package main // The backup holder's item events and its give-way to a restore of the recovery module (design 47 G §554, §556, §563): each item // of a module is said started and ended on the seat node-backup; for the items of the holder of the seat mesh-broker the holder // says its start, then reads the recovery module's restore-started and restored once more, and when a restore runs it ends the // item put off, naming that restore's act, and starts it again on that act's restored. Said means acknowledged. import ( "context" "errors" "fmt" "sort" "strings" "sync" ) // The seat node-backup's events of an item, and the seat of the holder whose items are the bus's. const ( EventItemStarted = "item-started" EventItemEnded = "item-ended" BrokerSeat = "mesh-broker" ) // The item's outcomes on item-ended. const ( ItemDone = "done" ItemFailed = "failed" ItemPutOff = "put-off" ) // StoppedWhy is the error of an item said started and never ended, ended at the holder's next start. const StoppedWhy = "stopped" // ItemRef is one item of a module on a node. type ItemRef struct{ Module, Item string } // ItemBus is what the backup holder says and reads on the bus. type ItemBus interface { // Say publishes one item event, `item-started` or `item-ended`, and answers once the bus acknowledged it. Say(event string, ref ItemRef, body map[string]any) error // Unended are this node's items said started whose last start is newer than their last end, read last per subject (§556). Unended() ([]ItemRef, error) // Restoring reads the recovery module's restore-started and restored of a module's kept data once more, last per subject, and // answers the act of a restore that runs, "" when none does. Restoring(module string) (string, error) // WaitRestored answers once the recovery module said restored with the act. WaitRestored(ctx context.Context, act string) error } // ItemsOf are the items of a module a backup covers: the data items it declares that are not a cache. func (b *Backups) ItemsOf(module string) []ItemRef { all, err := b.Items() if err != nil { b.Say("the data items of %s could not be read: %v", module, err) return nil } var out []ItemRef for _, it := range all[module] { if it.Class == "cache" { continue } out = append(out, ItemRef{Module: module, Item: it.ID}) } return out } func (b *Backups) sayItem(event string, ref ItemRef, body map[string]any) error { body["at"] = b.Now().UTC() return b.Bus.Say(event, ref, body) } // StopStale ends every item this node said started and never ended with `failed` and the error `stopped`, so no restore waits on // a dump that no longer runs (§563). It holds the night's lock, so no backup of this module is running meanwhile. func (b *Backups) StopStale() error { if b.Bus == nil { return nil } release, err := b.hold(nightLock) if err != nil { return err } defer release() left, err := b.Bus.Unended() if err != nil { return fmt.Errorf("the items said started and never ended could not be read: %w", err) } var failed []string for _, ref := range left { if err := b.sayItem(EventItemEnded, ref, map[string]any{"outcome": ItemFailed, "error": StoppedWhy}); err != nil { failed = append(failed, ref.Module+"."+ref.Item+": "+err.Error()) } } if len(failed) > 0 { return errors.New("items left started could not be ended: " + strings.Join(failed, "; ")) } return nil } // beginItems says every item of a module started and, for the items of the holder of the seat mesh-broker, gives way to a restore // that runs: its items are ended put off naming the act, wait for that restore's restored, and are started again. It answers how to // end the items when the backup ends. A module with no item says nothing. func (b *Backups) beginItems(ctx context.Context, module string) (func(err error), error) { nothing := func(error) {} if b.Bus == nil { return nothing, nil } items := b.ItemsOf(module) if len(items) == 0 { return nothing, nil } started := func() error { for _, ref := range items { if err := b.sayItem(EventItemStarted, ref, map[string]any{}); err != nil { return err } } return nil } end := func(outcome string, extra map[string]any) error { var first error for _, ref := range items { body := map[string]any{"outcome": outcome} for k, v := range extra { body[k] = v } if err := b.sayItem(EventItemEnded, ref, body); err != nil && first == nil { first = err } } return first } finish := func(err error) { var said error if err != nil { said = end(ItemFailed, map[string]any{"error": err.Error()}) } else { said = end(ItemDone, nil) } if said != nil { b.Say("the end of %s's items could not be said: %v", module, said) } } if err := started(); err != nil { return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) } b.clearWaits(items) if b.BrokerModule == nil { return finish, nil } broker, err := b.BrokerModule() if err != nil { _ = end(ItemFailed, map[string]any{"error": err.Error()}) return nothing, fmt.Errorf("who holds the seat %s is not known: %w", BrokerSeat, err) } if broker != module { return finish, nil } for { act, err := b.Bus.Restoring(module) if err != nil { _ = end(ItemFailed, map[string]any{"error": err.Error()}) return nothing, fmt.Errorf("the recovery module's restores could not be read: %w", err) } if act == "" { return finish, nil } b.Say("%s's backup gives way to the restore %s of the recovery module", module, act) if err := end(ItemPutOff, map[string]any{"restore": act}); err != nil { return nothing, fmt.Errorf("the put-off of %s's items could not be said: %w", module, err) } wctx, stop := context.WithCancel(ctx) var expiredMu sync.Mutex expired := false release, err := b.keepPutOff(items, act, func(ref ItemRef) { expiredMu.Lock() expired = true expiredMu.Unlock() b.writeWaits(ref, act) stop() }) if err != nil { stop() return nothing, err } waitErr := b.Bus.WaitRestored(wctx, act) release() stop() expiredMu.Lock() ran := expired expiredMu.Unlock() if ran { return nothing, fmt.Errorf("%w: the restore %s", errPutOffExpired, act) } if waitErr != nil { return nothing, fmt.Errorf("the end of the restore %s could not be waited for: %w", act, waitErr) } if err := started(); err != nil { return nothing, fmt.Errorf("the start of %s's items could not be said: %w", module, err) } b.clearWaits(items) } } // sortRefs orders items by module, then item. func sortRefs(refs []ItemRef) { sort.Slice(refs, func(i, j int) bool { if refs[i].Module != refs[j].Module { return refs[i].Module < refs[j].Module } return refs[i].Item < refs[j].Item }) }