Documentation
¶
Overview ¶
Package ops manages the async queue of long-running file operations (copy/move/delete) so the HTTP request can return promptly while the actual byte movement happens in a worker goroutine.
Persistence is in the pending_ops table — restart-safe so a server crash doesn't lose in-flight work. The single goroutine in Run() polls every few seconds, picks the oldest queued row, and executes it via the configured storage driver.
Index ¶
- Constants
- Variables
- func MoveDest(ctx context.Context, drv storage.Driver, src, dst string, taken ...Taken) (string, error)
- func SayTrashBatch(lang, kind string, queued bool, done, failed int, reason, name string) string
- func SayTrashEmpty(lang string, op *Op) string
- func SayTrashEmptyPreview(lang string, count int, bytes int64) string
- func Targets(kind string, sources []string, dest string) (src, dst []writegate.Target)
- func TenantKey(ctx context.Context) string
- func UniqueDest(ctx context.Context, drv storage.Driver, dst string, taken ...Taken) (string, error)
- func UniqueDestNumbered(ctx context.Context, drv storage.Driver, dst string, taken ...Taken) (string, error)
- func WithEncryptionSettled(ctx context.Context, sources []string) context.Context
- type DBSync
- type Job
- type Op
- type OpOpen
- type OpOutput
- type PluginRunner
- type Purger
- type RenameSync
- type RestoreFencer
- type Restorer
- type Service
- func (s *Service) Cancel(ctx context.Context, id int64) (bool, error)
- func (s *Service) Get(ctx context.Context, id int64) (*Op, error)
- func (s *Service) LatestThumbRepair(ctx context.Context, key string) (*Op, error)
- func (s *Service) LatestTrashEmpty(ctx context.Context, key string) (*Op, error)
- func (s *Service) List(ctx context.Context, status string) ([]*Op, error)
- func (s *Service) ListFor(ctx context.Context, status string, v Viewer) ([]*Op, error)
- func (s *Service) ListIn(ctx context.Context, status string, storageIDs []int64) ([]*Op, error)
- func (s *Service) Migrate(ctx context.Context) error
- func (s *Service) Run(ctx context.Context)
- func (s *Service) SetDecorator(fn func(ctx context.Context, rows []*Op))
- func (s *Service) SetDeleteWorkers(n int)
- func (s *Service) SetLocks(fn func(ctx context.Context, storageID int64) writegate.Locks)
- func (s *Service) SetPluginRunner(r PluginRunner)
- func (s *Service) SetPurger(p Purger)
- func (s *Service) SetRestorer(r Restorer)
- func (s *Service) SetSync(d DBSync)
- func (s *Service) SetThumbRepairer(r ThumbRepairer)
- func (s *Service) SetTrashEmptier(e TrashEmptier)
- func (s *Service) SetUploadCommitter(c UploadCommitter)
- func (s *Service) Stop()
- func (s *Service) Submit(ctx context.Context, kind string, storageID int64, sources []string, ...) (*Op, error)
- func (s *Service) SubmitJob(ctx context.Context, kind string, storageID int64, sources []string, ...) (*Op, error)
- func (s *Service) SubmitJobWithCleanup(ctx context.Context, kind string, storageID int64, sources []string, ...) (*Op, error)
- func (s *Service) SubmitThumbRepair(ctx context.Context, req ThumbRepairRequest) (*Op, error)
- func (s *Service) SubmitTo(ctx context.Context, kind string, storageID, destStorageID int64, ...) (*Op, error)
- func (s *Service) SubmitTrashEmpty(ctx context.Context, req TrashEmptyRequest) (*Op, error)
- func (s *Service) ThumbRepairDone(id int64) <-chan struct{}
- func (s *Service) TrashEmptyDone(id int64) <-chan struct{}
- type Skipped
- type SkipsError
- type Taken
- type ThumbRefusal
- type ThumbRepairCounts
- type ThumbRepairJob
- type ThumbRepairRequest
- type ThumbRepairer
- type TransferHooks
- type TrashEmptier
- type TrashEmptyRequest
- type UploadCommitter
- type Viewer
Constants ¶
const ( // SkipCycle — a folder link whose target this walk has already carried // once (storage.CycleGuard). Its content is at the destination, reached // through the first link; walking it again would only nest copies. SkipCycle = "cycle" // SkipTooDeep — a folder more than storage.MaxWalkDepth levels below the // one being carried. ⚠ This one CAN hold real data nobody linked, which // is the reason a move keeps its source when anything was skipped. SkipTooDeep = "too_deep" )
Reasons a transfer gives that are not a driver's link state.
const ( // OpRename gives ONE item a new name within its storage: Sources holds it // and Dest the exact path it becomes. It is not a move. A person chose // the name, so a destination that is taken fails the op (ErrNameTaken) // instead of landing beside it as `name-copy`, and nothing is replaced. OpRename = "rename" // OpRestore brings trash entries back where they came from. Sources holds // their node ids; the Restorer does the work, the same call // POST /api/files/manager/restore makes. OpRestore = "restore" // OpPurge takes trash entries out for good. Sources holds their node ids; // the Purger does the work, the same call DELETE /api/admin/trash/{id} // makes. It ran inside that request, and a folder is purged one object and // one row at a time: the admin page's client gave up after 30 s. OpPurge = "purge" )
const ( // ReasonExists: something already holds the entry's original place // (a restore; nothing moved, the entry is still in the trash). ReasonExists = "exists" // ReasonNotFound: the entry is not in the trash (any more), or is not the // caller's to see. ReasonNotFound = "not_found" // ReasonForbidden: the caller may not do this to the entry. ReasonForbidden = "forbidden" // ReasonFailed: anything else. ReasonFailed = "failed" )
Why the first entry of a trash batch that did not go did not go: the `reason_code` of a batch answer ({done, failed, reason_code}).
const ( OpCopy = "copy" OpMove = "move" OpDelete = "delete" // OpArchiveCreate materializes and compresses an archive through a handler- // supplied in-memory job. The persisted row intentionally contains only // safe metadata; archive passwords must never be written to pending_ops. OpArchiveCreate = "archive-create" // OpArchiveExtract expands an archive through a handler-supplied in-memory // job. As with creation, credentials stay in the closure and are never // persisted in pending_ops. OpArchiveExtract = "archive-extract" // OpUploadCommit streams a staged upload from filex's staging area into the // storage driver. Its single "source" is the staged upload id, not a path — // the work is done by the injected UploadCommitter, because the DB mirror, // the search index, thumbnails and the writehook all live in the handler // layer (same reason DBSync is injected rather than reimplemented here). OpUploadCommit = "upload-commit" // OpPluginAction runs one app-plugin action (internal/wasmplugin). The // row's Dest carries the job id (app_plugin_jobs) and Sources the input // paths; the work is done by the injected PluginRunner, which owns the // sandbox, the output commit and the job row. One op = one job, so // Done/Failed move as a unit rather than per source. OpPluginAction = "plugin-action" )
Op kinds.
const ( StatusPending = "pending" StatusRunning = "running" StatusOK = "ok" StatusFailed = "failed" StatusPartial = "partial" StatusCancelling = "cancelling" // StatusCancelled is a row Cancel ended: a pending row that never ran, // or a running one whose context was cancelled. StatusCancelled = "cancelled" )
Status values.
const ( ThumbRepairFix = "fix" ThumbRepairRebuild = "rebuild" )
Repair modes: Fix draws what is missing, failed, skipped or stale; Rebuild draws everything in scope.
const DefaultDeleteWorkers = 4
DefaultDeleteWorkers is the pool size when nobody sets one (config: FILEX_OPS_DELETE_WORKERS).
const OpThumbRepair = "thumb-repair"
OpThumbRepair is "repair thumbnails" (see the file comment).
const OpTrashEmpty = "trash-empty"
OpTrashEmpty is "empty the trash now" (see the file comment).
const TakenPrefix = "something already exists at this path: "
TakenPrefix starts the error a restore job records for an entry whose original place is taken (handlers.Trash.RestoreNode); the rest is the name. The row's summary reads the reason and the name back from it.
const TrashPrefix = trash.Prefix
TrashPrefix is the in-storage dir soft-deleted files are moved into. Re-exported from the trash package so there is exactly one definition of where the trash lives — the key minting itself is trash.Put's job now.
Variables ¶
var ErrEncryptedNameTaken = fmt.Errorf("%w: an item with this name already exists in this encrypted folder, and the server cannot choose another name for it", ErrNoFreeName)
ErrEncryptedNameTaken means the destination, inside an end-to-end encrypted folder whose names are encrypted, already holds an item with this name — and the server, holding no key, cannot pick another name that would still decrypt. Nothing was written. Wraps ErrNoFreeName, so every caller that answers "name taken" for that answers it for this too.
var ErrEncryptionUnasked = errors.New("the source could not be confirmed as checked against the end-to-end encryption policy when it was queued, and a file under this name is a new encryption; queue it again")
ErrEncryptionUnasked fails a source that is a file now, that the rule's names do not free, and that the queue cannot confirm the handler checked when it queued the work: it was a folder then (a folder lands on the name unasked), or the server has restarted since and the queue was told nothing. Under that name a file is a new end-to-end encryption.
var ErrIntoOwnDescendant = errors.New("ops: a folder cannot be moved or copied into itself or into one of its own subfolders")
ErrIntoOwnDescendant is "you asked to put this folder inside itself".
⚠ Measured on a local storage 2026-09-13, before this guard existed: the submit answered 202, the row went to the queue, and the WORKER failed a moment later with `rename /srv/data/x /srv/data/x/child/x: invalid argument`. Three things wrong with that. The person is told nothing at click time; the reason they eventually see is an OS errno and a SERVER path, which is both meaningless to them and more than they should be shown; and the refusal is the local driver's, not ours — an object store has no rename, so the same request there walks the tree and copies it into a destination that is inside the tree it is walking.
The answer belongs here, synchronously, in the one funnel both the unified endpoint and the per-verb wrappers pass through. A destination picker can (and does) grey the folder out, but a picker is a courtesy: the API is reachable without it.
var ErrNameTaken = errors.New("something with that name already exists here")
ErrNameTaken is a rename refused because something already has the name. Nothing moved.
var ErrNoFreeName = errors.New("no free name left beside the destination")
ErrNoFreeName means every name a move or copy would give an item beside a taken destination is taken as well. Nothing was written.
var ErrThumbRepairBusy = errors.New("ops: thumbnails are already being repaired")
ErrThumbRepairBusy is SubmitThumbRepair's answer while the caller's tenant already has a repair queued or running; the op returned beside it is that one.
var ErrTrashBusy = errors.New("ops: the trash is already being emptied")
ErrTrashBusy is SubmitTrashEmpty's answer while the caller's tenant already has an empty queued or running; the op returned beside it is that one.
Functions ¶
func MoveDest ¶
func MoveDest(ctx context.Context, drv storage.Driver, src, dst string, taken ...Taken) (string, error)
MoveDest is where a same-storage move of `src` to `dst` lands: `dst` when nothing holds it, a free name beside it when something does, and `src` itself when the move would put the item back where it already is.
⚠⚠ The one rule for every same-storage move — the queued worker and the manager's synchronous `?action=move` both call it. A driver's Move onto an occupied path REPLACES what is there (a local rename does, and so does an object store's copy-then-delete), and neither caller checked: measured 2026-09-14, moving `a.txt` into a folder holding an `a.txt` finished `ok` and the file that had been there was gone, not even in the trash. A copy and a cross-storage move already kept both; this makes the same-storage move agree with them rather than invent a third behaviour.
func SayTrashBatch ¶ added in v0.54.0
SayTrashBatch is what a batch of trash entries restored (kind OpRestore) or deleted for good (OpPurge) says, in lang: how many went - or, queued, are on their way - and how many did not and why (reason, one of the Reason codes; name is the entry a taken place names). One part, or both joined.
func SayTrashEmpty ¶ added in v0.54.0
SayTrashEmpty is where an "empty the trash" run stands, in lang: waiting its turn, running (so far of all), or how it ended. ⚠ It never sends a person to the server log: what could not be deleted is still in the trash, and that is what they can act on.
func SayTrashEmptyPreview ¶ added in v0.54.0
SayTrashEmptyPreview is the question "Empty the trash?" asks: what the purge would delete, counted the way the purge counts (trash.Service.Tally).
func Targets ¶
Targets is what one queued operation writes, for writegate.Check: on the source storage and on the destination storage. SubmitTo judges every operation by it — the explorer's copy, move and delete, the unified /api/files/ops endpoint, an app's action and a scheduled one — and the handlers ask the same function earlier, so the two cannot disagree about what an operation touches.
- move and delete TAKE their sources (and everything under a folder): a frozen document, or the folder around it, does not move or go.
- copy and an app's action only READ their sources: named, not changed. (An app's output is judged where it is written, handlers.AppPlugins.)
- a copy or move destination is named, not replaced: the worker de-collides (UniqueDest / MoveDest), so nothing already there is overwritten.
- a rename TAKES its source and WRITES its new name, the way the explorer's own rename is judged; a taken name fails the op.
- a restore's and a purge's sources are trash entry ids, not paths: the handler judges each entry before it queues them (handlers.Trash).
Every one of them is judged on filex's own names — a copy INTO `.filex-trash` is a file nobody can find, a move OUT of `.filex-open` takes a document from under the desktop that is about to write it back. Upload commits are not judged here: their sources are staged-upload ids, and the target was judged when the upload began (StagedUpload.Begin).
func TenantKey ¶
TenantKey names whose "empty the trash" a caller's is: its tenant, or "" for a caller that reaches every storage (single-tenant, the supertenant). A run is shown to, followed by and refused to the callers with the same key.
func UniqueDest ¶
func UniqueDest(ctx context.Context, drv storage.Driver, dst string, taken ...Taken) (string, error)
UniqueDest resolves a non-colliding destination on `drv` — `name-copy`, `name-copy-2`, … — so a paste never overwrites what is already there. It fails with ErrNoFreeName rather than ever returning a taken name; taken adds checks the driver cannot make (see Taken).
func UniqueDestNumbered ¶ added in v0.48.0
func UniqueDestNumbered(ctx context.Context, drv storage.Driver, dst string, taken ...Taken) (string, error)
UniqueDestNumbered is UniqueDest for a second DOCUMENT rather than a copy: `dst` when nothing holds it, else `name (2).ext`, `name (3).ext`, … — the numbering the New document dialog suggests for a name already taken. A draft saved beside a file of its name lands there (issue #71). Same rules, same Taken checks, same ErrNoFreeName as UniqueDest; only the spelling differs.
func WithEncryptionSettled ¶ added in v0.51.0
WithEncryptionSettled is ctx for the Submit or SubmitTo of a rename, a move or a copy onto a literal destination: sources are the ones the handler settled with the encryption rule (asked about and allowed, or free by their names). A source left out may still land as a folder, never as a file.
Types ¶
type DBSync ¶
type DBSync interface {
// SyncMove updates the moved node's path/parent in the DB.
SyncMove(ctx context.Context, storageID int64, src, dst string)
// SyncSoftDelete flags the node deleted and retags it to the trash path
// (storage_key keeps the original path so Restore works).
SyncSoftDelete(ctx context.Context, storageID int64, src, trashRel string)
// SyncHardDelete drops the node's rows (and everything below a folder)
// when its bytes are gone for good: the driver could not move it to the
// trash and deleted it outright, it was already missing, or a move across
// storages took it away. Nothing of it is listed in the trash.
SyncHardDelete(ctx context.Context, storageID int64, src string)
// SyncCopy inserts a DB node for the freshly written copy.
SyncCopy(ctx context.Context, storageID int64, src, dst string)
// SyncCopyAcross inserts a DB node for a copy whose two ends live in
// DIFFERENT storages. SyncCopy cannot serve this: it reads the source
// node and writes the new one under ONE storage id, so a cross-storage
// paste mirrored through it would either find no source node or attach
// the copy to the wrong depo — the listing would then show the file
// where it is not.
SyncCopyAcross(ctx context.Context, srcStorageID int64, src string, dstStorageID int64, dst string)
}
DBSync mirrors a completed filesystem operation into the DB node index. Implemented by the manager HTTP handler and injected via SetSync once both are constructed.
Without it the worker moves/deletes bytes on disk but leaves the DB cache stale. Directory listings read the DB (Store.ListNodesByParent), so a move would keep showing the file in its old folder and a delete would keep showing the file at all — the exact "move/delete doesn't work" bug. It also lets delete go through the trash (soft-delete) instead of hard-deleting.
type Job ¶ added in v0.44.2
Job is handler-owned background work. progress reports completed units; callers choose the units when submitting (archive creation uses one per staged member plus one for compression and the destination write).
type Op ¶
type Op struct {
ID int64 `json:"id"`
Kind string `json:"kind"`
StorageID int64 `json:"storage_id"`
// DestStorageID is the storage the destination lives in. It equals
// StorageID for the ordinary same-storage copy/move; a different value is
// what makes an op CROSS-STORAGE (paste from one depo into another),
// which the worker serves by streaming bytes between the two drivers.
// Zero means "same as StorageID" — the shape every row written before
// this column existed has.
DestStorageID int64 `json:"dest_storage_id,omitempty"`
Sources []string `json:"sources"`
Dest string `json:"dest,omitempty"`
Total int `json:"total"`
Done int `json:"done"`
// BytesTotal / BytesDone are a running cross-storage transfer's byte
// counters (issue #27), merged in from memory by Get/List — never stored.
// BytesTotal 0 with BytesDone > 0 means the total is not known (yet).
BytesTotal int64 `json:"bytes_total,omitempty"`
BytesDone int64 `json:"bytes_done,omitempty"`
// ObjectsTotal / ObjectsDone count the objects a running same-storage
// copy, move or delete has found and finished inside its sources: one
// folder on an object store is one source and minutes of objects, which
// the source counts above cannot show. From the storage driver's tally
// (storage.Tally); live counters like the bytes, never stored.
ObjectsTotal int64 `json:"objects_total,omitempty"`
ObjectsDone int64 `json:"objects_done,omitempty"`
Failed int `json:"failed"`
// Skipped counts items the op deliberately left alone (migration 00075):
// a thumbnail repair's files no engine draws, or end-to-end encrypted.
Skipped int `json:"skipped,omitempty"`
Status string `json:"status"`
Error string `json:"error,omitempty"`
// ErrorCode / ErrorEngine classify a failed APP job for the client, which
// says it in the person's language (lib/errorWords `jobFailure`) and keeps
// `Error` — English, sometimes plumbing — for an administrator's second
// line. Filled by the plugin Decorator, never stored.
ErrorCode string `json:"error_code,omitempty"`
ErrorEngine string `json:"error_engine,omitempty"`
// ErrorParams are the values ErrorCode's sentence takes, kept on the row
// with the code (errcode.go); ErrorText is that sentence in the reader's
// language, made when the row is read (sayErrors) - what every client
// prints. Error stays the English detail.
ErrorParams map[string]string `json:"error_params,omitempty"`
ErrorText string `json:"error_text,omitempty"`
CreatedAt time.Time `json:"created_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
// Plugin fields are filled by the Decorator for OpPluginAction rows and
// never stored here: which plugin/action ran, the action's label in the
// caller's locale, the last progress message, and the committed outputs.
Plugin string `json:"plugin,omitempty"`
Action string `json:"action,omitempty"`
Label string `json:"label,omitempty"`
Message string `json:"message,omitempty"`
Outputs []OpOutput `json:"outputs,omitempty"`
// Open is where a FINISHED plugin job sends the person who queued it: one
// of its outputs (adapter-qualified) and the app's own screen to start
// there - the job result's `surface.open` (wasmplugin openAfter). The
// client that queued the job follows it; nobody else's does. Filled by the
// Decorator, never stored here.
Open *OpOpen `json:"open,omitempty"`
// ActorID is who asked for this op. The worker runs on a server-lifetime
// context long after the request that queued the work is gone, so the only
// way a pasted file can be attributed to the person who pasted it is for
// the queue row to carry them (migration 00038). nil is SYSTEM — a row
// queued before the column existed, or by something that is not a person.
ActorID *int64 `json:"actor_id,omitempty"`
// Cancellable is advertised rather than inferred from the operation kind.
Cancellable bool `json:"cancellable"`
// Summary is where a trash job stands, said in the reader's language
// (say.go): a restore, a permanent delete or "empty the trash". Filled by
// Get/List for those kinds, never stored.
Summary string `json:"summary,omitempty"`
// contains filtered or unexported fields
}
Op is a queued file operation row.
func (*Op) ThumbRepairOf ¶ added in v0.50.0
func (op *Op) ThumbRepairOf() (job ThumbRepairJob, refused []ThumbRefusal, tenantKey string, ok bool)
ThumbRepairOf reads an OpThumbRepair row's ask and outcome back. ok is false for any other kind of row.
type OpOpen ¶ added in v0.50.0
type OpOpen struct {
Path string `json:"path"`
Action string `json:"action,omitempty"`
View string `json:"view,omitempty"`
}
OpOpen is "go to this file, and start that screen on it", for the person whose job produced the file - the same request a screen's `open` makes (wire.OpenRequest). Action and View are the app's own; naming neither just opens the file.
type OpOutput ¶
type OpOutput struct {
Path string `json:"path"`
}
OpOutput is one file a plugin job produced.
type PluginRunner ¶
type PluginRunner interface {
RunPluginAction(ctx context.Context, op *Op, live func(done, total int64)) error
}
PluginRunner executes an OpPluginAction row: reads the job behind op.Dest, runs the plugin in its sandbox, commits the outputs and records the job's outcome. live receives progress the plugin reports (done/total), which the tray draws through the same byte counters a cross-storage copy uses. Implemented by wasmplugin.Registry, injected via SetPluginRunner.
type Purger ¶ added in v0.47.0
Purger takes one trash entry out for good: its bytes, its rows and its descendants'. handlers.Trash implements it; wired with SetPurger. storageID is the job's, as for Restorer.
type RenameSync ¶ added in v0.47.0
type RenameSync interface {
NameTaken(ctx context.Context, storageID int64, src, dst string) (bool, error)
SyncRename(ctx context.Context, storageID int64, src, dst string)
}
RenameSync is what a DBSync adds to serve a rename (handlers.Manager does): the one rule for whether a name is taken — the catalogue's live rows as well as the storage, and a change of case alone is not the item colliding with itself — and the catalogue's side of the rename, announced as a rename.
type RestoreFencer ¶ added in v0.55.0
type RestoreFencer interface {
RestoreFence(ctx context.Context, storageID, nodeID int64) []string
}
RestoreFencer is what a Restorer may also answer: the prefixes a restore of nodeID fences instead of holding the storage's row gate (a folder an object store brings back one object at a time; nil: the gate). handlers.Trash answers it from the trash entry's row.
type Restorer ¶ added in v0.47.0
Restorer brings one trash entry back: the bytes to their original path and the row out of the trash. handlers.Trash implements it; wired with SetRestorer. An entry whose place is taken is an error, and nothing of it moves.
storageID is the storage the job was queued for (Op.StorageID): an entry that lives in another one is refused, not restored. The handler that queued the job judged its entries on that storage, and a row is the only thing the worker has to go on.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is the queue + worker bundle.
func New ¶
New returns a Service that talks to the given *sql.DB, assuming SQLite.
Callers must invoke Migrate before Submit/Status. Run starts the worker goroutine.
func NewForDialect ¶
func NewForDialect(database *sql.DB, dialect string, resolver func(int64) (storage.Driver, error)) *Service
NewForDialect is New for a server whose database is not SQLite.
⚠ Pass the real driver name. This queue is raw SQL on the application's own connection, not a db.Store, so nothing else in the process knows which dialect it is talking to: on PostgreSQL a Service built with the wrong dialect accepts every copy, move and delete and then fails each one at the first placeholder — which is precisely how issue #19's install behaved.
func (*Service) Cancel ¶
Cancel ends an op. A pending row is marked cancelled without running; a running row has its context cancelled, which tears down the plugin instance (or stops the transfer loop), and is marked cancelled when the worker returns. Unknown or finished rows are a no-op with ok=false.
func (*Service) LatestThumbRepair ¶ added in v0.50.0
LatestThumbRepair is the tenant's newest repair, in any state, or nil when it has asked for none.
func (*Service) LatestTrashEmpty ¶
LatestTrashEmpty is the tenant's newest "empty the trash", in any state, or nil when it has asked for none.
func (*Service) List ¶
List returns ops, optionally filtered by status (e.g. "running"). Empty status returns the most-recent rows across all statuses, capped at 200 to keep the polling payload small. Used by the SPA's PendingOpsTray which calls GET /api/files/ops?status=running every 2s.
func (*Service) ListIn ¶
ListIn is List restricted to a set of storage ids — the multi-tenant form. A nil set means "no restriction" (single-tenant mode, the supertenant, and every background caller), which is why List delegates here rather than the other way round.
⚠ The predicate is in the SQL, not applied to List's result, and that is the whole point of the extra method: the 200-row cap is applied by the database. Filtering afterwards would hand a tenant an EMPTY queue tray whenever another tenant had 200 more recent ops — isolation that silently costs the neighbour their own feature. A row matches on either end, because a cross-storage copy belongs to the tenant on either side of it.
func (*Service) Migrate ¶
Migrate prepares the queue for this boot.
⚠ It no longer creates pending_ops. The table is a goose migration like every other (db/migrations/*/00036_pending_ops.sql), because the hand-rolled CREATE TABLE that used to live here was SQLite DDL — INTEGER PRIMARY KEY AUTOINCREMENT — executed against whatever engine the operator had. On PostgreSQL it failed on every boot, the table never existed, and every copy, move and delete died on "relation pending_ops does not exist" while the server reported itself healthy (issue #19).
func (*Service) Run ¶
Run blocks until ctx is cancelled, draining the queue continuously.
Spawn it in its own goroutine — typically once at server boot.
func (*Service) SetDecorator ¶
SetDecorator installs the hook Get/List call to fill the plugin fields of rows (Op.Plugin, Label, Outputs, …) from the job table.
func (*Service) SetDeleteWorkers ¶
SetDeleteWorkers sets how many items of one delete job are trashed at the same time. Below 1 means 1. Call before Run.
func (*Service) SetLocks ¶
SetLocks wires the app-lock lookup SubmitTo judges every operation against (see Targets).
func (*Service) SetPluginRunner ¶
func (s *Service) SetPluginRunner(r PluginRunner)
SetPluginRunner wires the app-plugin executor. Without it an OpPluginAction row fails loudly.
func (*Service) SetPurger ¶ added in v0.47.0
SetPurger wires the purge behind an OpPurge row. Without it the row fails rather than claiming to have purged anything.
func (*Service) SetRestorer ¶ added in v0.47.0
SetRestorer wires the restore behind an OpRestore row. Without it the row fails rather than claiming to have restored anything.
func (*Service) SetThumbRepairer ¶ added in v0.50.0
func (s *Service) SetThumbRepairer(r ThumbRepairer)
SetThumbRepairer wires the repair. Without it SubmitThumbRepair refuses.
func (*Service) SetTrashEmptier ¶
func (s *Service) SetTrashEmptier(e TrashEmptier)
SetTrashEmptier wires the purge. Without it SubmitTrashEmpty refuses.
func (*Service) SetUploadCommitter ¶
func (s *Service) SetUploadCommitter(c UploadCommitter)
SetUploadCommitter wires the staged-upload transfer hook. Call once at boot, before Run.
func (*Service) Stop ¶
func (s *Service) Stop()
Stop signals the worker to exit and waits for it.
⚠ The finishing lane is waited for at most finishGrace. A rename, a restore or a purge runs its storage work to the end whatever happens around it (rename_restore.go), and a large folder on an object store is minutes of it: Stop used to wait for all of it. Past the grace the job carries on in the background for as long as the process does; its row stays `running` until the job writes how it ended, and a row still `running` at the next boot is requeued (Migrate) and carried on (runRename resumes).
func (*Service) Submit ¶
func (s *Service) Submit(ctx context.Context, kind string, storageID int64, sources []string, dest string) (*Op, error)
Submit enqueues a same-storage op and pokes the worker.
Kept as the narrow form because most callers (delete, upload-commit, and every copy/move that stays inside one storage) have nothing to say about a destination storage. It delegates to SubmitTo with dest == source.
func (*Service) SubmitJob ¶ added in v0.44.2
func (s *Service) SubmitJob(ctx context.Context, kind string, storageID int64, sources []string, dest string, total int, job Job) (*Op, error)
SubmitJob queues handler-owned work without persisting its executable payload. This is important for encrypted archives: sources and destination are useful durable operation metadata, but the password stays exclusively in the closure and therefore in process memory.
func (*Service) SubmitJobWithCleanup ¶ added in v0.44.2
func (s *Service) SubmitJobWithCleanup(ctx context.Context, kind string, storageID int64, sources []string, dest string, total int, job Job, cleanup func()) (*Op, error)
SubmitJobWithCleanup is SubmitJob with an idempotent cleanup callback. The callback is also invoked when a still-pending job is cancelled, which is essential for request-prepared temporary inputs that the job never gets a chance to defer-remove itself.
func (*Service) SubmitThumbRepair ¶ added in v0.50.0
SubmitThumbRepair queues a repair and starts it.
While the caller's tenant already has one queued or running, nothing is queued: the answer is ErrThumbRepairBusy with that run. Another tenant's run does not refuse this one.
func (*Service) SubmitTo ¶
func (s *Service) SubmitTo(ctx context.Context, kind string, storageID, destStorageID int64, sources []string, dest string) (*Op, error)
SubmitTo enqueues an op whose destination may live in ANOTHER storage.
destStorageID == storageID (or 0) is the ordinary same-storage op. A different id makes the worker stream the bytes from one driver to the other — the "copy from one depo, paste into the next" gesture, which used to be accepted and then silently written into the SOURCE storage because the queue had nowhere to put the target's storage.
func (*Service) SubmitTrashEmpty ¶
SubmitTrashEmpty queues "empty the trash now" and starts it.
While the caller's tenant already has one queued or running, nothing is queued: the answer is ErrTrashBusy with that run. A run of ANOTHER tenant (or of the platform, or the nightly retention sweep) does not refuse this one — it waits its turn behind it, queued, and says so.
func (*Service) ThumbRepairDone ¶ added in v0.50.0
ThumbRepairDone is closed when the run behind op id has ended, or nil when this process is not running it.
func (*Service) TrashEmptyDone ¶
TrashEmptyDone is closed when the run behind op id has ended, or nil when this process is not running it.
type Skipped ¶
type Skipped struct {
// Path is the entry's source path, storage-relative.
Path string `json:"path"`
// Reason is the source driver's own link state (storage.LinkBroken,
// LinkOutsideRoot, LinkUnresolved), or SkipCycle / SkipTooDeep.
Reason string `json:"reason"`
}
Skipped is one entry inside a transferred folder that was deliberately NOT carried, and why.
⚠⚠ Why a transfer skips at all, measured 2026-09-21 against this code before the rule existed (a `local` source, follow_symlinks off, copying a folder holding one link of each kind to a second storage):
broken link the WHOLE copy failed: read "tree/broken": not found
link out of the root the WHOLE copy failed: symlink target is outside…
`cycle -> .` 41 nested duplicate folders written into the
destination, THEN failed: too many levels of links
unresolved link, on a "ok" — and the bytes of the file the link pointed
source whose Read at, outside the storage, sitting in the destination
follows it (sftp, ftp)
The local driver's Read refuses an out-of-root link, which is why the second row failed instead of leaking; a remote driver's Read is the SERVER opening the path, and a server follows links wherever its account can read. So the rule cannot be "Read will say no": a link the driver did not resolve is never opened here at all, whatever the driver would do with it.
Skip, don't fail — the rule the same-storage folder copy already follows (local.Driver.copyTree): one link nobody can follow must not cost the other nine hundred and ninety-nine files. Unlike that copy, this one SAYS what it left behind — the op row ends `partial` with the list (Service.execute), and a move keeps its source (crossTransfer).
func Transfer ¶
func Transfer(ctx context.Context, srcDrv, dstDrv storage.Driver, src, dst string, hooks TransferHooks) ([]Skipped, error)
Transfer copies `src` (a file or a whole directory) from one storage driver to `dst` on another, verifying every file on arrival. What it deliberately did not carry comes back in the list (see Skipped); an error means the transfer itself failed.
It does not delete anything: a move is this call followed by the caller's own delete, which is the order that makes a failed transfer harmless — and a caller that deletes must not do so while the list is non-empty.
type SkipsError ¶
type SkipsError struct {
Skipped []Skipped
// SourceKept is set by a MOVE that did not delete its source because of
// the entries above (crossTransfer).
SourceKept bool
}
SkipsError is a transfer that finished and left entries behind. It is an error so that a caller cannot mistake it for a clean success by ignoring it; errors.As tells it apart from a failure.
func (*SkipsError) Error ¶
func (e *SkipsError) Error() string
type Taken ¶
Taken is an extra "is this name taken?" a caller can add to a destination check — the catalogue's live rows, which the driver cannot see when a row's bytes went missing. Moving onto such a name would make the catalogue drop that row, and its history with it.
type ThumbRefusal ¶ added in v0.50.0
ThumbRefusal is one storage a run refused, and the reason's code (server.Gap*: its catalogue cannot hold its files yet).
type ThumbRepairCounts ¶ added in v0.50.0
type ThumbRepairCounts struct {
Processed int
OK int
Failed int
Skipped int
// Refused are the storages the run would not repair, with why.
Refused []ThumbRefusal
}
ThumbRepairCounts is what a run has done so far.
type ThumbRepairJob ¶ added in v0.50.0
type ThumbRepairJob struct {
// StorageID is the storage; 0 is every storage in Reach.
StorageID int64
// Path is a file, or a folder with everything in it, inside StorageID:
// storage-relative. "" is the whole storage.
Path string
// Mode is ThumbRepairFix or ThumbRepairRebuild.
Mode string
// Reach is the storages the asker may repair: nil every one, empty none.
Reach []int64
}
ThumbRepairJob is what one run was asked to repair.
type ThumbRepairRequest ¶ added in v0.50.0
type ThumbRepairRequest struct {
Job ThumbRepairJob
// Tenant is TenantKey of the caller.
Tenant string
}
ThumbRepairRequest is what an admin asked to repair, and whose ask it is.
type ThumbRepairer ¶ added in v0.50.0
type ThumbRepairer interface {
// CountRepair is how many files the job would draw right now.
CountRepair(ctx context.Context, job ThumbRepairJob) (int, error)
// RunRepair draws them, calling progress as it goes.
RunRepair(ctx context.Context, job ThumbRepairJob, progress func(ThumbRepairCounts)) (ThumbRepairCounts, error)
}
ThumbRepairer runs the repair behind an OpThumbRepair row: the server's thumbnail walk. Injected with SetThumbRepairer.
type TransferHooks ¶
type TransferHooks struct {
OnDir func(src, dst string)
OnFile func(src, dst string, size int64)
// OnBytes is told how many new bytes were read from a source file, while
// they stream (issue #27). Unlike the two above it fires before
// verification — it measures motion, not completion.
OnBytes func(n int64)
}
TransferHooks let the caller mirror each finished step into whatever catalogue it keeps. Both are optional and are called only after the bytes are on the far side and verified.
This exists so the agent/MCP surface and the queue can share ONE transfer engine. A second implementation of "walk a tree between two drivers" would drift — and the half that drifts is always the one nobody watches.
type TrashEmptier ¶
type TrashEmptier interface {
Tally(ctx context.Context, job trash.EmptyJob) (count int, bytes int64, err error)
RunEmpty(ctx context.Context, job trash.EmptyJob, started func() bool, progress func(trash.PurgeResult)) (trash.PurgeResult, error)
}
TrashEmptier runs the purge behind an OpTrashEmpty row. *trash.Service implements it; injected with SetTrashEmptier.
type TrashEmptyRequest ¶
type TrashEmptyRequest struct {
// StorageID narrows the run to one storage; 0 is every storage in Reach.
StorageID int64
// OlderThanDays: 0 is everything that is in the trash when it is asked.
OlderThanDays int
// Reach is the storages the caller may empty (trash.Reach): nil every one.
Reach []int64
// Tenant is TenantKey of the caller: whose run it is.
Tenant string
}
TrashEmptyRequest is what an admin asked to empty.
type UploadCommitter ¶
UploadCommitter transfers one staged upload's bytes to the storage driver and runs the post-write side effects (node update, search index, thumbnail, writehook). Implemented by the staged-upload HTTP handler and injected via SetUploadCommitter once both are constructed.
Without it an OpUploadCommit op fails loudly rather than silently doing nothing: a staged upload whose bytes never move is data the user believes is saved.
type Viewer ¶
type Viewer struct {
All bool
StorageIDs []int64
Tenant string
// Own narrows the viewer to the rows Actor queued. A row carries its
// sources and its destination: paths inside folders the viewer may hold
// no grant for, which the tenant scope above does not look at. Set for
// everybody but an administrator (the handlers' opsViewer); a row that
// names nobody (queued before actor_id existed, or by something that is
// not a person) is then nobody's, and Actor 0 reaches no row at all.
Own bool
Actor int64
}
Viewer is who is looking at the queue: every row (All), or the rows of these storages plus the trash empties of this tenant — narrowed, when Own is set, to the rows Actor queued.
func ViewerOf ¶
ViewerOf is the viewer a request is: unscoped callers and the supertenant see every row; a tenant its storages' rows and its own trash empties.
func (Viewer) Sees ¶
Sees reports whether v may see (and so follow or cancel) op. An op is a tenant's when either of its storages is; a trash empty, which may name no storage at all, is the tenant's that asked for it. An Own viewer sees only what it queued (the same predicate ListFor puts in the SQL).