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 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)
- type DBSync
- type Op
- type OpOutput
- type PluginRunner
- 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) 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) SetSync(d DBSync)
- 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) 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) TrashEmptyDone(id int64) <-chan struct{}
- type Skipped
- type SkipsError
- type Taken
- 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 ( OpCopy = "copy" OpMove = "move" OpDelete = "delete" // 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" // 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 DefaultDeleteWorkers = 4
DefaultDeleteWorkers is the pool size when nobody sets one (config: FILEX_OPS_DELETE_WORKERS).
const OpTrashEmpty = "trash-empty"
OpTrashEmpty is "empty the trash now" (see the file comment).
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 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 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 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 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.
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).
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 flags the node deleted when the driver could not move it
// to trash and had to delete the bytes outright.
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 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"`
Failed int `json:"failed"`
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"`
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"`
// 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"`
// contains filtered or unexported fields
}
Op is a queued file operation row.
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 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) 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) 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) 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) 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) 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 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.
Source Files