Documentation
¶
Overview ¶
Package sync implements the storage-to-DB sync worker.
Each enabled storage gets one supervisor goroutine that picks the right strategy (poll vs fsnotify) and drives a per-run tombstone-guarded reconciliation against the storage backend.
Index ¶
- Constants
- Variables
- func CatalogueTree(ctx context.Context, store db.Store, idx *search.Index, ...) error
- func CountParts(etag string) int
- func MultipartETag(r io.Reader, partSize int64) (string, error)
- func RecomputeFolderSizes(ctx context.Context, store db.Store, storageID int64) error
- func ScopePath(raw string) (string, error)
- type ChangeEmitter
- type FolderRescan
- type SizeRefresher
- type Worker
- func (w *Worker) AddStorage(ctx context.Context, st *model.Storage) error
- func (w *Worker) AttachAntivirus(fn func(ctx context.Context, n *model.Node))
- func (w *Worker) AttachIndex(idx *search.Index)
- func (w *Worker) Known(storageID int64) bool
- func (w *Worker) RemoveStorage(id int64)
- func (w *Worker) RescanFolder(ctx context.Context, storageID int64, dir string) (FolderRescan, error)
- func (w *Worker) Running(storageID int64) bool
- func (w *Worker) Start(ctx context.Context) error
- func (w *Worker) Stop()
- func (w *Worker) Trigger(ctx context.Context, storageID int64) error
Constants ¶
const AbortedAtStartup = "interrupted: the server stopped during the scan"
AbortedAtStartup is the error recorded on a sync_runs row the worker closes at start because the process that opened it is gone.
const DefaultPollInterval = 15 * time.Minute
DefaultPollInterval is the cadence a polled storage falls back to when its own sync_interval_s is unset (or under the 5 s floor). FILEX_SYNC_INTERVAL overrides it; see NewWithInterval.
const FailureReportThreshold = 3
FailureReportThreshold is how many runs in a row must fail before a failure is reported as a WARNING (and so reaches the error tracker) rather than noted at INFO.
Why a single failed run is not an error ¶
A poll run reads the backend's listing. Object stores answer a transient 503/504 under load, and when the retry budget is spent the run gives up — but nothing is lost: the catalogue is simply not refreshed until the next tick, minutes later, which normally succeeds.
Measured on fm.example.com: "sync: run failed … ListObjectsV2 … 504" fired 15 times in six weeks against Hetzner Object Storage, every one of them followed by a successful run. Reporting each hiccup buys nothing and costs the thing that matters — an error tracker where a real outage stands out. Three in a row is roughly 45 minutes of a storage genuinely not answering, which IS worth waking up for.
Variables ¶
var ( // ErrScopeInvalid refuses a folder a rescan cannot be pointed at: one that // climbs out with "..", or one of filex's own trees. ErrScopeInvalid = errors.New("sync: not a folder that can be rescanned") // ErrFolderNotCatalogued is a folder the catalogue has no row for. A // rescan starts from the folder's row; rescan the folder above it, or run // a full scan. ErrFolderNotCatalogued = errors.New("sync: the folder is not in the catalogue") // ErrNotAFolder is a path whose row is a file. ErrNotAFolder = errors.New("sync: the path is a file, not a folder") )
var ErrRunInProgress = errors.New("sync: a run is already in progress for this storage")
ErrRunInProgress is what RunOnce (and so Worker.Trigger) returns when this storage is already being walked. It is not a failure of the run — the run the caller wanted is the one in progress — and noteRun does not count it.
var TreePrefetchMax = 2_000_000
TreePrefetchMax bounds how many objects one RunOnce will hold in memory from a single-pass listing before it gives up on the shortcut and walks directory by directory instead. Two million objects is on the order of a few hundred MB, which a server syncing a bucket that size has.
Functions ¶
func CatalogueTree ¶
func CatalogueTree(ctx context.Context, store db.Store, idx *search.Index, avScan func(ctx context.Context, n *model.Node), st *model.Storage, drv storage.Driver, dir string, parent *int64) error
CatalogueTree catalogues everything under dir on drv exactly as a sync pass catalogues a directory: rows the index does not have yet are created (with the driver's own size, mime, etag and mtime), indexed and handed to avScan; rows it already has are left as the walk leaves them. parent is the row of dir itself.
It exists for writes that put a whole subtree on the storage in ONE driver call — a same-storage folder copy — and have no per-file signal to mirror from. Reusing the walk rather than a second catalogue loop is the point: the walk's rules (the trash skip, the encrypted-folder marker ordering, the stale-row repair) cannot drift between the two.
⚠ Unlike RunOnce it does NOT strip the owner from ctx: the rows are billed to whoever ctx says wrote them.
func CountParts ¶
CountParts extracts the part count from an `<md5>-<count>` etag, or 1 for single-part etags.
func MultipartETag ¶
MultipartETag computes the S3-style multipart ETag of an io.Reader. `partSize` should match the upload chunk size (default 8MB).
The format is `<md5_of_concatenated_part_md5s>-<part_count>`.
func RecomputeFolderSizes ¶
RecomputeFolderSizes walks a storage's cached node tree and stores, for every folder, two aggregates in that folder's own row:
- size = the sum of its descendant file sizes (recursive total)
- backend_mtime = its newest descendant mtime ("last activity")
The file explorer then shows folder size + date served straight from the index — no backend re-scan, no per-folder query. The date matters for drivers that don't report a native directory mtime (e.g. synthetic S3 prefixes have none); "newest descendant" is both universal and the more useful semantic (it tracks when the folder's contents last changed).
Soft-deleted nodes (including trashed items under .filex-trash) are excluded by AggNodes, so trash never counts toward a folder's aggregates. Only rows whose value actually changed are written back. Called at the end of every sync (and can be invoked directly, e.g. after an upgrade, to backfill).
func ScopePath ¶
ScopePath normalises the folder a rescan was asked for, as a storage path with a leading slash. "" — for "", "/" and anything that cleans to the root — means the whole storage, i.e. an ordinary full scan.
".." is refused rather than cleaned: a request that climbs is not the folder it names, whatever it resolves to. So is anything inside filex's own trees (syspath.Sealed), which the walk never catalogues.
Types ¶
type ChangeEmitter ¶
type ChangeEmitter interface {
EmitChange(storageID int64, dir string, ev realtime.ChangeEvent)
}
ChangeEmitter is the realtime surface the refresher publishes through. The hub (*realtime.Hub) satisfies it.
type FolderRescan ¶
type FolderRescan struct {
Path string `json:"path"`
Scanned int `json:"scanned"`
Added int `json:"added"`
Updated int `json:"updated"`
Removed int `json:"removed"`
Reconciled int `json:"reconciled"`
// RemovalSkipped says why nothing was moved to the trash when nothing
// could be: part of the listing failed, or the 70% guard tripped.
RemovalSkipped string `json:"removal_skipped,omitempty"`
}
FolderRescan is what one folder rescan did (Worker.RescanFolder).
type SizeRefresher ¶
type SizeRefresher struct {
// contains filtered or unexported fields
}
SizeRefresher recomputes a storage's folder aggregates (RecomputeFolderSizes) shortly after its catalogue changed, instead of waiting for the next sync.
Issue #27: after a move the source and destination folders kept their old sizes. Folder sizes were only ever recomputed at the end of a sync pass, and a move, an upload or a delete through the explorer — or through WebDAV, S3, SFTP, NFS — changes the catalogue rows without touching those totals. On a storage that syncs rarely (or manually) the wrong size stood for hours.
Every mutation already announces itself to open explorers through the realtime emitter, so the refresher sits in front of that emitter (Wrap) and is told about every change in one place, whatever surface made it.
⚠ Debounced WITH a ceiling: a pure trailing debounce never fires while a stream keeps arriving (filex lesson #84) — a 5000-file upload would show the old sizes until its very end. `quiet` is the gap that ends a burst, `maxWait` the longest a burst may postpone a recompute.
⚠ The listing refresh the refresher sends goes to the INNER emitter, never through Wrap: otherwise every recompute would schedule the next one.
func NewSizeRefresher ¶
func NewSizeRefresher(recompute func(ctx context.Context, storageID int64) error, emit ChangeEmitter, quiet, maxWait time.Duration) *SizeRefresher
NewSizeRefresher builds a refresher over recompute (normally RecomputeFolderSizes bound to the store). emit may be nil (no listing refresh is sent then).
func (*SizeRefresher) Touch ¶
func (r *SizeRefresher) Touch(storageID int64, dir string)
Touch records that dir in storageID changed and schedules a recompute.
func (*SizeRefresher) Wrap ¶
func (r *SizeRefresher) Wrap(inner ChangeEmitter) ChangeEmitter
Wrap returns an emitter that forwards every change to inner and tells the refresher about it. Install the wrapped emitter wherever mutations publish.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker is the top-level sync supervisor — one instance per server.
func NewWithInterval ¶
NewWithInterval is New with an explicit global fallback cadence — what FILEX_SYNC_INTERVAL sets.
⚠ Fallback ONLY: a storage row with its own sync_interval_s still wins. That is exactly what docs/CONFIGURATION.md has always claimed the variable did. It did not — the value was parsed into config and then read by nothing, while the real fallback was a hardcoded 15m literal in loopPoll that happened to be the same number, which is why the dead knob was invisible.
func (*Worker) AddStorage ¶
AddStorage launches a syncer for a newly-created storage row.
func (*Worker) AttachAntivirus ¶
AttachAntivirus wires the antivirus enqueue so files that arrive ON a storage — rather than through a filex write surface — are scanned too.
A file dropped into a bucket with `aws s3 cp`, written on a mounted disk by another process, or simply already there when the storage was added, is discovered by the walk, catalogued and content-indexed. Until this hook it was never handed to ClamAV, which is precisely the place an operator who turned scanning on assumes a scan happens.
fn is queue.AntivirusScanner.EnqueueDiscovered bound to the queue driver: it is best-effort by contract (an enqueue failure is logged, never returned) and it applies Supports()/Eligible() itself, so directories, oversized files and anything under `.filex-trash/` or `.versions/` are refused there rather than second-guessed here. nil (no ClamAV binary, or no persistent queue) leaves the walk byte for byte as it was.
func (*Worker) AttachIndex ¶
AttachIndex wires a Bleve index into the worker so each sync upsert also lands as a search document. Without this, search only knows about whatever the admin's `Rebuild` button has flushed.
func (*Worker) Known ¶
Known reports whether a storage has a registered syncer — what Trigger would otherwise only discover after the caller had stopped listening.
func (*Worker) RemoveStorage ¶
RemoveStorage stops the syncer for a deleted storage.
func (*Worker) RescanFolder ¶
func (w *Worker) RescanFolder(ctx context.Context, storageID int64, dir string) (FolderRescan, error)
RescanFolder rescans one catalogued folder's subtree of a storage instead of the whole storage.
A full scan is the only other way to make the catalogue look at a storage again, and it scales with the storage: 169k rows took 23 minutes and rewrote seen_at on every one of them, to learn about one folder. This runs the same walk over the folder alone — the same drift update, the same staged-upload settle, the same tombstone rules — with three differences that keep it a folder's business:
- the tombstone candidates are bounded by the folder's exact prefix, and the 70% guard compares what the listing saw with the folder's OWN live row count, not the storage's last run;
- a listing that failed part-way removes nothing (see walkCounts.partial);
- no sync_runs row, and the storage's last-synced time does not move: a folder rescan is not a sync of the storage.
It shares the storage's one-run lock: ErrRunInProgress while any run walks.
func (*Worker) Start ¶
Start launches one syncer per enabled storage. ctx is the parent shutdown context.
⚠ First it closes every sync_runs row still open. A run records its own end, and a process that stopped in the middle of one — restarted, killed, out of memory — never did: its row said `running` for ever, on panels, to the thumbnail backfill's "is a sync running?" check, and in the history. Nothing of this process can be running yet, so an open row here belongs to one that is gone.
This is Start's job and not server.New's on purpose: `filex thumb backfill` builds a whole Server beside a live one (it never calls Start), and closing rows there would declare the live server's scan dead while it walks.
Source Files