mesh-catalog/modules/restic/cmd/restic-backups/items.go

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
	})
}