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 ObjectDrift(n *model.Node, obj storage.Object) bool
- func RecomputeFolderSizes(ctx context.Context, store db.Store, storageID int64) error
- func ScopePath(raw string) (string, error)
- type CatalogueCoverage
- 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) AttachEmitter(e ChangeEmitter)
- func (w *Worker) AttachIndex(idx *search.Index)
- func (w *Worker) AttachPairRoots(f func(storageID int64) []string)
- func (w *Worker) CatalogueCoverage(ctx context.Context, st *model.Storage) *CatalogueCoverage
- func (w *Worker) CatalogueCurrent(storageID int64, dir string) (lazy, current bool)
- func (w *Worker) CatalogueOpened(storageID int64, dir string)
- func (w *Worker) CatalogueSizePartial(ctx context.Context, storageID int64, dirs []string) map[string]bool
- func (w *Worker) CatalogueStatus(ctx context.Context, storageID int64) (*CatalogueCoverage, bool)
- func (w *Worker) CatalogueSubtree(storageID int64, dir string)
- func (w *Worker) Known(storageID int64) bool
- func (w *Worker) NoteActivity(storageID int64)
- 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 ( // CoverageFilling: a lazy storage whose background filler has not // converged yet (behaviour A). CoverageFilling = "lazy_filling" // CoverageVisitedOnly: a lazy storage in behaviour B — only the folders // people open are catalogued. CoverageVisitedOnly = "lazy_on_open" // CoverageFirstScan: any storage whose first full scan has not finished. CoverageFirstScan = "first_scan" )
Coverage reasons: why a storage's catalogue is not complete.
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 ( // LazyFillIdlePause is the filler's breath between two folders while // nobody is using the storage. LazyFillIdlePause = 2 * time.Millisecond // LazyFillBusyPause is the pause while somebody is: every listing, search // or open of the storage in the last LazyActiveWindow slows the filler to // this, and every open request runs before its next folder regardless. LazyFillBusyPause = 250 * time.Millisecond // LazyActiveWindow is how long one request counts as "somebody is here". LazyActiveWindow = 15 * time.Second // LazyWatchQuiet ends a burst of fsnotify events for a folder; // LazyWatchMaxWait is the longest a burst may postpone its reconcile. LazyWatchQuiet = 2 * time.Second LazyWatchMaxWait = 10 * time.Second // LazySizeCheckpoint is how often folder sizes are recomputed while the // filler works (it does not announce each folder, so nothing else would). LazySizeCheckpoint = 30 * time.Second // LazyFillBatch is how many frontier rows the filler reads at a time. LazyFillBatch = 64 // LazyPairRefreshFloor is the shortest interval at which the subtrees of // desktop sync pairs are walked again (pairLoop); the storage's own sync // interval is used when it is longer. LazyPairRefreshFloor = 30 * time.Second )
Tunables. Package variables rather than constants so the tests can run the filler and the watch debounce at test speed; nothing else changes them.
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 ObjectDrift ¶ added in v0.44.2
ObjectDrift is objectDrift for the explorer's listing: a folder listed from the disk with the catalogue overlaid (handlers/lazy_listing.go) shows the disk's size and date for an entry whose row has drifted by exactly the rule the scan updates rows by — so the listing and the next reconcile agree.
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 CatalogueCoverage ¶ added in v0.44.2
type CatalogueCoverage struct {
// Complete: search, folder sizes and usage cover the whole storage.
Complete bool `json:"complete"`
// Reason says why not (CoverageFilling, CoverageVisitedOnly,
// CoverageFirstScan); empty when complete.
Reason string `json:"reason,omitempty"`
// Mode is "lazy" for a lazily catalogued storage, empty otherwise.
Mode string `json:"mode,omitempty"`
// Fill is the lazy behaviour: "background" (A) or "on_open" (B).
Fill string `json:"fill,omitempty"`
// CataloguedFolders / PendingFolders: folders whose listing has been
// applied, and folders discovered but not listed yet.
CataloguedFolders int64 `json:"catalogued_folders"`
PendingFolders int64 `json:"pending_folders"`
// WatchedFolders / MaxWatches: the watch budget, in use and in total.
WatchedFolders int `json:"watched_folders,omitempty"`
MaxWatches int `json:"max_watches,omitempty"`
// Filler is the background filler's state: filling, paused (somebody is
// using the storage), idle, converged, refreshing, or off (behaviour B).
Filler string `json:"filler,omitempty"`
// HeldBackFolders: folders whose last reconcile refused deletions.
HeldBackFolders int64 `json:"held_back_folders,omitempty"`
// Reconciles since the syncer started.
Reconciles int64 `json:"reconciles,omitempty"`
}
CatalogueCoverage is how much of a storage its catalogue covers — the object the listing, search and drive-usage responses carry, and the admin storage page's catalogue block.
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) AttachEmitter ¶ added in v0.44.2
func (w *Worker) AttachEmitter(e ChangeEmitter)
AttachEmitter wires the realtime chain the lazy catalogue announces its findings through (handlers' emitter: change log → size refresher → hub). Safe to call after Start: it is read at announce time.
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) AttachPairRoots ¶ added in v0.44.2
AttachPairRoots wires where the lazy catalogue learns which folders desktop sync pairs mirror right now (realtime.Hub.WatchedRoots: the roots of every live recursive watch on a storage). Read at refresh time, like the emitter.
func (*Worker) CatalogueCoverage ¶ added in v0.44.2
CatalogueCoverage describes how much of st its catalogue covers, or nil when it covers everything. A lazy storage answers from its per-folder state; any other storage is incomplete only while its first scan has not finished.
func (*Worker) CatalogueCurrent ¶ added in v0.44.2
CatalogueCurrent reports whether storageID is lazy and, if so, whether the catalogue vouches for dir right now (watched, nothing held back). The explorer lists a lazy folder that is not current from the disk.
func (*Worker) CatalogueOpened ¶ added in v0.44.2
CatalogueOpened tells a lazy storage that somebody listed dir: a watched, current folder only moves up the watch LRU; anything else is queued for a reconcile ahead of the background filler. Never blocks. A no-op on any other storage.
func (*Worker) CatalogueSizePartial ¶ added in v0.44.2
func (w *Worker) CatalogueSizePartial(ctx context.Context, storageID int64, dirs []string) map[string]bool
CatalogueSizePartial reports, for folders of a lazy storage, whether each one's catalogued size leaves something out (see lazyCatalogue.sizePartial). Keys are canonical folder paths ("/a/b"). Nil for any other storage.
func (*Worker) CatalogueStatus ¶ added in v0.44.2
CatalogueStatus is the admin storage page's catalogue block: the lazy engine's full state, complete or not. ok is false for a storage that is not running lazily.
func (*Worker) CatalogueSubtree ¶ added in v0.44.2
CatalogueSubtree makes sure every folder under dir of a lazy storage gets catalogued — a desktop sync pair's subtree (see lazyCatalogue.subtree).
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) NoteActivity ¶ added in v0.44.2
NoteActivity tells a lazy storage's filler that somebody is using the storage (a search, say), so it slows down.
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