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.
- this package — DB-backed adapter that implements storage.FailureRecorder + storage.EventNotifier and also runs the reconciliation queue handler + 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 ¶
- func RetryDedupKey(path, op string) string
- type CronScheduler
- type FailureRecorder
- type Notifier
- type Reconciled
- type RulesReloader
- type Service
- func (s *Service) FixOne(ctx context.Context, path, op string) (queued bool, err error)
- func (s *Service) GenerateReport(ctx context.Context) error
- func (s *Service) HandleRetry(ctx context.Context, op queue.Op) error
- func (s *Service) ReconcileAll(ctx context.Context) (Reconciled, error)
- func (s *Service) Stop()
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
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.
func NewFailureRecorder ¶
func NewFailureRecorder(store db.Store) *FailureRecorder
NewFailureRecorder binds the recorder to a store.
type Notifier ¶
type Notifier struct {
// contains filtered or unexported fields
}
Notifier is the storage.EventNotifier impl backed by notify.Service.
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 + cron status report. Wires store, queue, notify and the live ReplicatedDriver.
The queue handler for op type "replica_retry" should be Service. HandleRetry — the bootstrap registers it on the queue Pool.
func New ¶
func New(store db.Store, driver *storage.ReplicatedDriver, q queue.Driver, n notify.Service) *Service
New wires a Service.
func (*Service) FixOne ¶
FixOne enqueues a single retry for one (path, op) pair, 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) HandleRetry ¶
HandleRetry is the queue.Handler for op type "replica_retry". The bootstrap calls Pool.Register(queue.TypeReplicaRetry, svc.HandleRetry).
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) 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.
Source Files