Documentation
¶
Overview ¶
Package replica is the orchestration layer that ties the storage ReplicatedDriver to the DB (failure rows, rules, settings) and to the notify subsystem (webhook + bell events).
The split is:
- internal/storage/replicated.go — the wrapper Driver itself.
- internal/storage/rules.go — the path-glob rule engine.
- internal/server storage_cache.go — the resolver's cache, which wraps a storage linked to an enabled replication target.
- this package — DB-backed adapter that implements storage.FailureRecorder + storage.EventNotifier and also runs the reconciliation queue handler, the initial copy and the cron status report.
Keeping the storage package pure of DB and notify imports lets us unit-test the wrapper in isolation against in-memory recorders.
Index ¶
- Variables
- func EnsureFolder(ctx context.Context, store db.Store, st *model.Storage, targetID int64) (string, error)
- func FolderName(name string, storageID int64) string
- func InitialCopyDedupKey(storageID int64) string
- func RetryDedupKey(storageID int64, path, op string) string
- type CronScheduler
- type FailureRecorder
- type InitialCopyStatus
- type Link
- type LinkStatus
- type Notifier
- type Reconciled
- type RulesReloader
- type Service
- func (s *Service) FixOne(ctx context.Context, storageID int64, path, op string) (queued bool, err error)
- func (s *Service) GenerateReport(ctx context.Context) error
- func (s *Service) HandleInitialCopy(ctx context.Context, op queue.Op) error
- func (s *Service) HandleRetry(ctx context.Context, op queue.Op) error
- func (s *Service) HasLinked(ctx context.Context) (bool, error)
- func (s *Service) InitialCopies(ctx context.Context) ([]InitialCopyStatus, error)
- func (s *Service) LinkChanged(ctx context.Context, storageID int64) error
- func (s *Service) Linked(ctx context.Context) ([]*model.Storage, error)
- func (s *Service) Links(ctx context.Context) ([]LinkStatus, error)
- func (s *Service) ReconcileAll(ctx context.Context) (Reconciled, error)
- func (s *Service) RestartInitialCopy(ctx context.Context, storageID int64) error
- func (s *Service) ResumeInitialCopies(ctx context.Context) error
- func (s *Service) SetClock(now func() time.Time)
- func (s *Service) SetFolder(ctx context.Context, storageID int64, folder string) (*model.ReplicaLink, error)
- func (s *Service) SetSlice(d time.Duration, files int)
- func (s *Service) StartInitialCopy(ctx context.Context, storageID, targetID int64, restart bool) (*model.ReplicaInitialCopy, error)
- func (s *Service) Stop()
- type Wrappers
Constants ¶
This section is empty.
Variables ¶
var ErrFolderTaken = errors.New("replica: another storage on this target already uses that folder")
ErrFolderTaken is SetFolder's answer for a folder another storage on the same target already writes into.
var ErrNoReplica = errors.New("no replica configured")
ErrNoReplica is what the repair calls answer when no storage replicates at all (the admin endpoints turn it into 503 "no replica configured").
var ErrNotLinked = errors.New("replica: storage is not linked to an enabled replication target")
ErrNotLinked is what Wrappers answers for a storage that does not replicate: it has no replication target, its target is switched off or gone, or the storage itself is gone.
Functions ¶
func EnsureFolder ¶ added in v0.53.0
func EnsureFolder(ctx context.Context, store db.Store, st *model.Storage, targetID int64) (string, error)
EnsureFolder returns the storage's folder on targetID, choosing and keeping one the first time: its name made safe (FolderName), or - when another storage on the same target already writes into that name - the name with the storage's id appended. A link kept for another target is replaced (the storage was relinked).
func FolderName ¶ added in v0.53.0
FolderName makes a folder name every backend accepts out of a storage's name: letters (any script) and digits stay, a space, `-`, `_` and `.` stay, everything else - `/ \ : * ? " < > |`, control characters - becomes `-`; runs of `-` are one, the ends lose dots, spaces and dashes (a leading dot would hide it, a trailing one Windows drops), and it is at most 64 characters. A name that ends up empty, or that Windows reserves (CON, LPT1...), becomes storage-<id>.
func InitialCopyDedupKey ¶ added in v0.53.0
InitialCopyDedupKey keeps at most one waiting slice per storage.
func RetryDedupKey ¶ added in v0.47.0
RetryDedupKey coalesces the retries of one failure: while a retry of it is waiting in the queue, another request for it adds nothing.
⚠ Every press of "Repair all" queued a full set again. The list looks the same until a retry has run, which invites another press, and each one doubled the queue and told the bell "retries queued" once more.
Types ¶
type CronScheduler ¶
type CronScheduler struct {
// contains filtered or unexported fields
}
CronScheduler wraps robfig/cron with a Reload primitive so the admin UI can change the cron spec at runtime without a restart.
func NewCronScheduler ¶
func NewCronScheduler(svc *Service) *CronScheduler
NewCronScheduler binds a scheduler to the replica Service.
func (*CronScheduler) Reload ¶
func (c *CronScheduler) Reload(ctx context.Context) error
Reload re-reads the replica_settings row and reschedules. Empty cron spec or report_enabled=false removes any active schedule.
Safe to call concurrently; the mutex serializes spec swaps so the scheduler never has two entries for the same job.
func (*CronScheduler) Start ¶
func (c *CronScheduler) Start()
Start launches the underlying cron loop. Must be called once before the first Reload — Reload is a no-op until Start.
func (*CronScheduler) Stop ¶
func (c *CronScheduler) Stop()
Stop blocks until any in-flight job returns and the scheduler tear down completes.
type FailureRecorder ¶
type FailureRecorder struct {
// contains filtered or unexported fields
}
FailureRecorder is a DB-backed storage.FailureRecorder for ONE storage: every row it writes or resolves carries that storage's id (migration 00094).
func NewFailureRecorder ¶
func NewFailureRecorder(store db.Store, storageID int64) *FailureRecorder
NewFailureRecorder binds the recorder to a store and a storage.
type InitialCopyStatus ¶ added in v0.53.0
type InitialCopyStatus struct {
*model.ReplicaInitialCopy
StorageName string `json:"storage_name"`
TargetName string `json:"target_name"`
// Done is Copied+Present+Excluded+Failed.
Done int64 `json:"done"`
}
InitialCopyStatus is a storage's initial copy as the Replication page shows it: the row, with the storage's and the target's names.
type Link ¶ added in v0.53.0
type Link struct {
StorageID int64
TargetID int64
Driver *storage.ReplicatedDriver
}
Link is a storage's live replication: the wrapper the running process writes through, and the target it fans out to.
type LinkStatus ¶ added in v0.53.0
type LinkStatus struct {
*model.ReplicaLink
StorageName string `json:"storage_name"`
TargetName string `json:"target_name"`
}
LinkStatus is a storage's link as the Replication page shows it.
type Notifier ¶
type Notifier struct {
// contains filtered or unexported fields
}
Notifier is the storage.EventNotifier impl backed by notify.Service.
The service is read when an event is sent, not when the notifier is made: the server builds the storages' wrappers before its notification service exists, and a wrapper made then must still reach the bell.
func NewNotifier ¶
NewNotifier wires the adapter to a service that already exists.
func NewNotifierFunc ¶ added in v0.53.0
NewNotifierFunc wires the adapter to a service read at send time.
func (*Notifier) ForStorage ¶ added in v0.53.0
ForStorage returns a copy that names the storage in what it sends.
func (*Notifier) NotifyPrimaryReadFail ¶
NotifyPrimaryReadFail emits a primary_read_fail event when the read fallback served from replica.
type Reconciled ¶ added in v0.47.0
Reconciled is what one "Repair all" did: retries it queued, and retries it found already waiting in the queue and left alone.
type RulesReloader ¶
type RulesReloader struct {
// contains filtered or unexported fields
}
RulesReloader is the helper an admin handler calls after rules CRUD.
func NewRulesEngine ¶
func NewRulesEngine(store db.Store) (storage.RuleEngine, *RulesReloader)
NewRulesEngine wires storage.NewRulesEngine with a reload function that pulls from db.Store. Call .Reload() after admin CRUD ops.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service orchestrates reconciliation, the initial copy and the cron status report. Wires store, queue, notify and the live wrappers.
The queue handler for op type "replica_retry" should be Service. HandleRetry — the bootstrap registers it on the queue Pool; the initial copy's is HandleInitialCopy (TypeInitialCopy).
func New ¶
New wires a Service. wrappers may be nil (tests of the report alone): then nothing can be repaired or copied and those calls answer ErrNoReplica.
func (*Service) FixOne ¶
func (s *Service) FixOne(ctx context.Context, storageID int64, path, op string) (queued bool, err error)
FixOne enqueues a single retry for one (storage, path, op), unless one is waiting in the queue already (queued=false, no error).
func (*Service) GenerateReport ¶
GenerateReport computes the singleton replica_status_reports row and sends the full event to the webhook (in-app gets a summary).
The full failed-paths list goes only to webhooks (the user posts it into their own system — F2/F3) — the in-app body keeps the message short so the bell doesn't drown in JSON.
func (*Service) HandleInitialCopy ¶ added in v0.53.0
HandleInitialCopy is the queue.Handler for TypeReplicaInitialCopy: one slice of one storage's initial copy.
func (*Service) HandleRetry ¶
HandleRetry is the queue.Handler for op type "replica_retry". The bootstrap calls Pool.Register(queue.TypeReplicaRetry, svc.HandleRetry).
The retry goes through the storage's LIVE wrapper (Wrappers), so it writes to the target the storage is linked to now. A storage that no longer replicates - unlinked, its target switched off or deleted, the storage gone - leaves nothing to repair: its failure is resolved and the op succeeds.
The handler is structured to never panic and always either:
- succeed (success → Ack on the queue side)
- return an error so the queue requeues with backoff
func (*Service) HasLinked ¶ added in v0.53.0
HasLinked reports whether any storage replicates. The repair endpoints answer 503 "no replica configured" only when none does.
func (*Service) InitialCopies ¶ added in v0.53.0
func (s *Service) InitialCopies(ctx context.Context) ([]InitialCopyStatus, error)
InitialCopies lists every storage's initial copy, for the page.
func (*Service) LinkChanged ¶ added in v0.53.0
LinkChanged brings the storage's initial copy in line with its link as the row says now: linked to an enabled target - a copy to that target (a new one when the target changed); not linked - none.
func (*Service) Linked ¶ added in v0.53.0
Linked returns the storages that replicate: linked to a replication target that exists and is enabled.
func (*Service) Links ¶ added in v0.53.0
func (s *Service) Links(ctx context.Context) ([]LinkStatus, error)
Links lists every storage's folder on its target.
func (*Service) ReconcileAll ¶
func (s *Service) ReconcileAll(ctx context.Context) (Reconciled, error)
ReconcileAll enqueues a replica_retry op for every unresolved failure currently recorded that has none waiting already.
func (*Service) RestartInitialCopy ¶ added in v0.53.0
RestartInitialCopy is LinkChanged that begins the copy anew even when one to the same target finished: the target's configuration changed (another bucket, another share), or it was switched back on after the changes made while it was off went nowhere.
func (*Service) ResumeInitialCopies ¶ added in v0.53.0
ResumeInitialCopies runs once at start: every storage that replicates gets its initial copy - a new one when it has none (the upgrade from a version whose replication never ran: the copy runs ONCE, for links that never had one) - and every copy still under way gets its next slice queued. Rows of storages that no longer replicate are dropped.
func (*Service) SetClock ¶ added in v0.53.0
SetClock replaces the clock the initial copy reads (tests).
func (*Service) SetFolder ¶ added in v0.53.0
func (s *Service) SetFolder(ctx context.Context, storageID int64, folder string) (*model.ReplicaLink, error)
SetFolder changes the storage's folder on its target to folder (made safe the way FolderName makes a name safe) and returns the one kept. Another storage on the same target writing into it is ErrFolderTaken. The caller rebuilds the storage's driver and begins its initial copy again; nothing is moved on the target.
func (*Service) SetSlice ¶ added in v0.53.0
SetSlice shortens or lengthens a slice (tests; files <= 0 means no file limit).
func (*Service) StartInitialCopy ¶ added in v0.53.0
func (s *Service) StartInitialCopy(ctx context.Context, storageID, targetID int64, restart bool) (*model.ReplicaInitialCopy, error)
StartInitialCopy makes sure the storage has an initial copy to targetID and queues its next slice. restart=false keeps a copy already there for that target (a finished one is not run again); restart=true begins it anew.
type Wrappers ¶ added in v0.53.0
type Wrappers interface {
// Replicated answers ErrNotLinked for a storage that does not replicate.
Replicated(ctx context.Context, storageID int64) (Link, error)
}
Wrappers finds the live replication wrapper of a storage - the one the server's resolver hands every writer (internal/server storage_cache.go).
⚠ Repairs and the initial copy go through THIS, not through a wrapper of their own: the Service was once built with a nil wrapper ("v0.1 skips that"), so every Fix all answered "no replica configured" whatever was linked (#186).
Source Files