Documentation
¶
Overview ¶
Package queue is filex' driver-based persistent operation queue.
Three drivers ship in the binary:
sqlite — default, single-node; uses the existing filex DB file postgres — production HA via SELECT FOR UPDATE SKIP LOCKED redis — priority-ordered sorted set, claimed with one Lua script
All drivers share the same Op struct and Driver contract so callers can swap them via the FILEMANAGER_QUEUE_DRIVER env var without touching handlers. Operations survive restarts (the queue table is the source of truth). On boot, drivers re-queue any rows left in `running` state from a previous crash.
The Pool type (worker.go) consumes a Driver and a map of handlers keyed by Op.Type. Handlers are pure Go functions; storage drivers, replica retry, reconcile and report jobs all attach as handlers.
Index ¶
- Constants
- Variables
- func AVDedupKey(nodeID int64) string
- func Names() []string
- func Register(name string, f Factory)
- type AVNodeStore
- type AVScanner
- type AntivirusScanner
- func (a *AntivirusScanner) AttachBody(b *filebody.Resolver)
- func (a *AntivirusScanner) AttachMaxBytes(fn func() int64)
- func (a *AntivirusScanner) Eligible(n *model.Node) bool
- func (a *AntivirusScanner) Enqueue(ctx context.Context, drv Driver, n *model.Node)
- func (a *AntivirusScanner) EnqueueAfterSave(ctx context.Context, drv Driver, n *model.Node, window time.Duration)
- func (a *AntivirusScanner) EnqueueDiscovered(ctx context.Context, drv Driver, n *model.Node)
- func (a *AntivirusScanner) Handle(ctx context.Context, op Op) error
- type ContentIndexer
- type Driver
- type Factory
- type Handler
- type NodeGetter
- type Op
- type Pool
- type PoolOption
- type Stats
Constants ¶
const ( StatusPending = "pending" StatusRunning = "running" StatusDone = "done" StatusFailed = "failed" StatusCancelled = "cancelled" )
Status values for Op.Status. Drivers must use exactly these strings — the admin UI renders them verbatim.
const ( TypeCopy = "copy" TypeMove = "move" TypeDelete = "delete" TypeReplicaRetry = "replica_retry" TypeReplicaReport = "replica_report" TypeReconcile = "reconcile" TypeThumb = "thumb" )
Common Op.Type values. Callers may invent new ones — drivers do not enforce a registry — but the canonical handlers below are wired in server bootstrap.
const DefaultContentMaxBytes int64 = 5 << 20
DefaultContentMaxBytes is the source-size ceiling for extraction when FILEX_SEARCH_CONTENT_MAX is unset (5 MiB — frozen v0.2 contract).
const DefaultMaxAttempts = 3
DefaultMaxAttempts is the retry budget when an Op enqueue request omits the field. Three attempts strikes a useful balance: transient network errors usually resolve by attempt 2; a third gives the upstream service a final chance before the op is parked in `failed` for the operator.
const PriorityDiscovered = -1
PriorityDiscovered is the Op.Priority given to a scan the STORAGE SYNC asked for, as opposed to one a person is waiting on.
⚠⚠ This exists because of one measurement. Point filex at a folder that already holds 20 000 files and every one of them is newly discovered, so one pass enqueues 20 000 scans. The queue orders `priority DESC, enqueued_at ASC`, so at equal priority those 20 000 rows are simply ahead of everything that arrives next — and an upload's scan, enqueued ten seconds into the backlog, waited **41 s** to be picked up: not most of a scan, most of the whole import. Dropping the sync's scans one step below everything else took the same probe to **1 ms**, because the only thing an interactive scan then waits for is a worker finishing the one scan it is holding.
Negative rather than "interactive positive" so that every op filex already enqueues — content extraction, thumbnails, replica retry, copy/move — keeps its current standing and also overtakes the sweep. Nothing else in filex sets Priority, so this is the first row that is not 0.
⚠ SQLite and Postgres order by it; the Redis driver's pending LIST is positional and ignores it. On Redis the backlog is FIFO and the 41 s is what an operator gets. That is a driver difference, not a policy: the fix belongs in the Redis driver (a second list, or a ZSET keyed by priority), and it is out of scope here.
const TypeAntivirusScan = "antivirus_scan"
TypeAntivirusScan is the op type for async ClamAV scanning ("Koru", v0.4): read a freshly written file from its storage driver, run the resolved clamdscan/clamscan binary over it, and — only when infected — quarantine the node into `.filex-trash/` + emit a `file.infected` event. Clean files see zero side effects, and the upload path is never blocked (content_index pattern: enqueue is best-effort, work happens on the worker pool).
const TypeContentIndex = "content_index"
TypeContentIndex is the op type for async content extraction ("Bul", v0.2): read an eligible file from its storage driver, extract plain text (internal/search/extract) and land it in the Bleve `content` field.
Variables ¶
var ErrDuplicate = errors.New("queue: duplicate op")
ErrDuplicate is returned by Enqueue when the op carries a DedupKey and a pending op already holds that key. It is a NORMAL outcome, not a failure: the caller asked for "one of these, please" and one already exists.
var ErrEmpty = errors.New("queue: empty")
ErrEmpty is returned by Dequeue when there is no eligible op.
var ErrNotFound = errors.New("queue: op not found")
ErrNotFound is returned when a specific op id can't be located.
Functions ¶
func AVDedupKey ¶
AVDedupKey is the coalescing key for a node's pending scan. One pending scan per node, which is the whole of the debounce rule: while an op holds this key, further requests for the same node are absorbed.
Types ¶
type AVNodeStore ¶
type AVNodeStore interface {
GetNode(ctx context.Context, id int64) (*model.Node, error)
SoftDeleteAndRetag(ctx context.Context, id int64, trashPath, trashHash, origPath string) error
}
AVNodeStore is the slim store contract for lookup + quarantine retag.
type AVScanner ¶
type AVScanner interface {
Supports() bool
Scan(ctx context.Context, r io.Reader) (infected bool, signature string, err error)
}
AVScanner is the slim scan contract the job needs — satisfied by *antivirus.Scanner, faked in tests.
type AntivirusScanner ¶
type AntivirusScanner struct {
// contains filtered or unexported fields
}
AntivirusScanner owns the antivirus_scan job. Enqueue fires from the upload surfaces (upload finalize, manager vfUpload, public drop); Handle runs on the worker pool.
func NewAntivirusScanner ¶
func NewAntivirusScanner(store AVNodeStore, resolver func(int64) (storage.Driver, error), sc AVScanner, n notify.Service, idx *search.Index, maxBytes int64) *AntivirusScanner
NewAntivirusScanner wires the job. maxBytes <= 0 disables the size gate override and callers should pass antivirus.MaxScanBytes().
func (*AntivirusScanner) AttachBody ¶
func (a *AntivirusScanner) AttachBody(b *filebody.Resolver)
AttachBody wires the byte-source resolver so a re-queued scan of a file that is still being transferred reads the staged bytes.
func (*AntivirusScanner) AttachMaxBytes ¶
func (a *AntivirusScanner) AttachMaxBytes(fn func() int64)
AttachMaxBytes makes the size ceiling live: fn is consulted per eligibility check. nil restores the fixed value passed to the constructor.
func (*AntivirusScanner) Eligible ¶
func (a *AntivirusScanner) Eligible(n *model.Node) bool
Eligible reports whether n qualifies for a scan: a live file within the size cap that is not itself a trash/version artifact.
func (*AntivirusScanner) Enqueue ¶
Enqueue schedules a scan for n when the scanner is available and n is eligible. Best-effort: enqueue failures are logged, never surfaced — a scan must not cost a write.
func (*AntivirusScanner) EnqueueAfterSave ¶
func (a *AntivirusScanner) EnqueueAfterSave(ctx context.Context, drv Driver, n *model.Node, window time.Duration)
EnqueueAfterSave schedules a DEBOUNCED scan for a file the browser's text editor has just overwritten: one scan, `window` from now, coalesced so that a burst of Ctrl+S costs exactly one scan.
Three properties make this work, and none of them is incidental:
- The delay is the queue's own not_before, never a time.AfterFunc. An in-process timer dies with the process and takes the pending scan with it, so every restart and every deploy would silently drop whatever was in flight — reintroducing, at exactly the moments a server is most likely to be restarted, the gap this feature exists to close. The wider the window, the more scans an in-process timer would lose.
- Repeat saves inside the window are DROPPED, not rescheduled. The window therefore starts at the first save and the scan is guaranteed to happen, rather than being pushed out indefinitely by someone who keeps typing.
- Handle resolves the node and opens its bytes at EXECUTION time, so the one scan that does run reads the file's final state, not the content of the save that scheduled it.
Best-effort, exactly like Enqueue: a scan must not cost a write. ErrDuplicate is the normal, expected outcome of the second save onwards and is not an error — it means the scan the caller wanted is already scheduled.
func (*AntivirusScanner) EnqueueDiscovered ¶
EnqueueDiscovered is Enqueue for a file the STORAGE SYNC found rather than a file somebody wrote: same op, same eligibility, same handler, one step down the queue. See PriorityDiscovered.
func (*AntivirusScanner) Handle ¶
func (a *AntivirusScanner) Handle(ctx context.Context, op Op) error
Handle processes one antivirus_scan op. Vanished/ineligible nodes resolve as done (nil). Storage-read and scan failures return an error so the queue's retry budget applies. A clean verdict has no side effects; an infected verdict quarantines the node (storage rename into `.filex-trash/` + DB soft-delete retag) and emits `file.infected`.
type ContentIndexer ¶
type ContentIndexer struct {
// contains filtered or unexported fields
}
ContentIndexer owns the content_index job: Enqueue is wired as the search index's content hook (fires on every metadata index whose content fingerprint drifted), Handle is registered on the worker pool.
Everything here is best-effort by design — the write path never blocks on extraction, and extraction failures are logged + skipped rather than failing anything user-visible. Only transport-level failures (storage read, index write) return an error so the queue's retry budget applies.
func NewContentIndexer ¶
func NewContentIndexer(store NodeGetter, resolver func(int64) (storage.Driver, error), idx *search.Index, maxBytes int64) *ContentIndexer
NewContentIndexer wires the job. maxBytes <= 0 falls back to DefaultContentMaxBytes.
func (*ContentIndexer) AttachBody ¶
func (c *ContentIndexer) AttachBody(b *filebody.Resolver)
AttachBody wires the byte-source resolver so a re-queued extraction of a file that is still being transferred reads the staged bytes.
func (*ContentIndexer) Eligible ¶
func (c *ContentIndexer) Eligible(n *model.Node) bool
Eligible reports whether n qualifies for content extraction: a live file within the size cap whose mime/extension has a registered extractor.
func (*ContentIndexer) Enqueue ¶
Enqueue schedules extraction for n when it is eligible. Best-effort: enqueue failures are logged, never surfaced — search freshness must not cost a write.
func (*ContentIndexer) Handle ¶
func (c *ContentIndexer) Handle(ctx context.Context, op Op) error
Handle processes one content_index op. Ineligible/vanished nodes resolve as done (nil); extraction failures are logged and the fingerprint is still recorded so the node doesn't re-enqueue forever; storage-read and index-write failures return an error to use the queue's retry budget.
type Driver ¶
type Driver interface {
// Init wires the driver to its backing store. Called once at boot.
Init(ctx context.Context, cfg map[string]any) error
// Name returns the registered driver name (e.g. "sqlite").
Name() string
// Enqueue persists op and returns its assigned id. Op.ID is ignored.
//
// When op.DedupKey is non-empty and a pending op already holds that
// key, implementations MUST return ErrDuplicate and persist nothing.
// Two concurrent Enqueues with the same key must resolve to exactly
// one stored op — never zero.
Enqueue(ctx context.Context, op Op) (id string, err error)
// Dequeue atomically picks and marks-running the next eligible op.
// types limits the candidate set — empty slice means any type.
// Returns ErrEmpty when nothing is ready (callers sleep + retry).
Dequeue(ctx context.Context, types []string) (Op, error)
// Ack marks the op completed. Done24h counters bump.
Ack(ctx context.Context, id string) error
// Fail records a failure. If retry is true, the op goes back to
// pending with attempts++; otherwise it terminates as failed.
Fail(ctx context.Context, id string, errMsg string, retry bool) error
// List returns ops filtered by status (or empty for any). Pagination
// via limit/offset; total returns the unpaginated count.
List(ctx context.Context, status string, limit, offset int) (ops []Op, total int64, err error)
// Get returns a single op by id. Returns ErrNotFound when missing.
Get(ctx context.Context, id string) (Op, error)
// Stats returns aggregated counters for the admin dashboard.
Stats(ctx context.Context) (Stats, error)
// Cancel transitions a pending op to cancelled. Running ops cannot
// be cancelled in v0.1 — the worker checks ctx.Err() but cooperative
// shutdown is on the caller's handler.
Cancel(ctx context.Context, id string) error
// Retry transitions a failed op back to pending (resetting attempts
// to attempts ≥ 0 — drivers may bump this if they wish). Used by
// admin UI's "retry" button.
Retry(ctx context.Context, id string) error
// RecoverOrphans flips long-running rows back to pending. Called on
// boot to handle ungraceful shutdowns. olderThan is the threshold
// (running rows whose started_at < now - olderThan re-queued).
RecoverOrphans(ctx context.Context, olderThan time.Duration) (int64, error)
// Close releases resources. Pool calls this on Stop.
Close() error
}
Driver is the persistence + dispatch contract for an op queue.
Implementations MUST be safe for concurrent calls — Pool spawns N worker goroutines, each calling Dequeue independently.
type Factory ¶
type Factory func() Driver
Factory constructs an empty Driver. Each call returns a fresh value so every queue instance has its own state. Drivers register one of these from their package init() block.
type Handler ¶
Handler is a function that processes a single Op and returns nil on success or an error to trigger retry/fail logic.
Handlers receive the parent ctx — they should respect cancellation so graceful shutdown completes within the configured timeout.
type NodeGetter ¶
NodeGetter is the slim store contract the content job needs.
type Op ¶
type Op struct {
ID string `json:"id"`
Type string `json:"type"`
Payload map[string]any `json:"payload"`
Status string `json:"status"`
Priority int `json:"priority"`
Attempts int `json:"attempts"`
MaxAttempts int `json:"max_attempts"`
LastError string `json:"last_error,omitempty"`
EnqueuedAt time.Time `json:"enqueued_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
NotBefore *time.Time `json:"not_before,omitempty"`
// DedupKey coalesces ops: while a PENDING op holds this key, another
// Enqueue with the same key is refused with ErrDuplicate instead of
// creating a second op. Empty (the default) opts out entirely, so every
// existing caller keeps its behaviour byte for byte.
//
// Scoped to `pending` on purpose. The key is released the moment the op
// is dequeued, so a request that arrives while the op is RUNNING queues
// a fresh one — the running scan may already have read the old bytes.
// The failure this ordering avoids is the dangerous one: dropping a
// request whose content nothing will ever look at. Queueing one scan too
// many wastes a scan; queueing none is the bug.
//
// ⚠ Write-only on the wire: drivers persist and enforce it, but Dequeue,
// Get and List do NOT populate it back. Assert on the ErrDuplicate
// behaviour, which is the actual contract, not on the field round-tripping.
DedupKey string `json:"-"`
}
Op is a single queued operation. Drivers persist this verbatim (with a JSON-encoded Payload).
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool runs N goroutines, each looping Dequeue → handler → Ack/Fail.
func NewPool ¶
func NewPool(drv Driver, workers int, opts ...PoolOption) *Pool
NewPool wires a worker pool with `workers` goroutines. workers must be ≥ 1; zero is silently corrected to 1 to avoid a stuck pool.
func (*Pool) Register ¶
Register attaches a handler for the given Op.Type. Re-registering overrides — useful in tests.
func (*Pool) Start ¶
Start launches the worker goroutines. Blocks the caller only briefly; goroutines run until Stop or ctx is cancelled.
type PoolOption ¶
type PoolOption func(*Pool)
PoolOption configures a Pool.
func WithLogger ¶
func WithLogger(l *slog.Logger) PoolOption
WithLogger plugs in a non-default slog.Logger.
func WithMaxBackoff ¶
func WithMaxBackoff(d time.Duration) PoolOption
WithMaxBackoff caps the exponential backoff used after errors.
func WithPollInterval ¶
func WithPollInterval(d time.Duration) PoolOption
WithPollInterval overrides the default 500ms poll cadence used when the queue is empty. Lower values reduce latency but increase DB load.
Directories
¶
| Path | Synopsis |
|---|---|
|
drivers
|
|
|
postgres
Package postgres is the production HA queue driver.
|
Package postgres is the production HA queue driver. |
|
redis
Package redis is the Redis-backed queue driver.
|
Package redis is the Redis-backed queue driver. |
|
sqlite
Package sqlite is the default queue driver, sharing the filex DB file with the metadata store.
|
Package sqlite is the default queue driver, sharing the filex DB file with the metadata store. |