jschoubben
c2710f52dd
The delivery repository's contracts and SDK live under modules/ as mesh-delivery-contracts and mesh-delivery-sdk, their Go module paths renamed to match the folders (the TypeScript names already did), and every importer in mesh-delivery, mesh-catalog, mesh-host, mesh-tools and mesh-sdk follows in the same group: a design as it should be, one path per module, checked by the group's check building every importer. Builds: design 47 G §427, §88 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
367 lines
16 KiB
Go
367 lines
16 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
|
"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
|
|
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"github.com/novox/mesh-host/internal/compositions"
|
|
)
|
|
|
|
// A module's kept data on the bus, taken as a restore point before a form change moves it and restored at a revert (design 47 G
|
|
// §296, §297, §554, §555, §560, §630 to §632, the lead's answers A-77, A-98 and A-108): the node agent asks the seat
|
|
// `mesh-recovery`'s `snapshot-kept` and waits for the `ended` of its own act, or, at a revert, stops the newer build, asks
|
|
// `kept-points`, asks `restore-kept` with the snapshot of the point of the form it reverts from, waits for its `restored`, and
|
|
// after the reverted build's apply calls the point's catch-up as the module's tool.
|
|
|
|
// Ending is the ending event of an act of the recovery module as the node agent reads it: `ended` for a restore point taken,
|
|
// `restored` for one restored (design 47 G §560).
|
|
type Ending struct {
|
|
Act string
|
|
Outcome string
|
|
Error string
|
|
Snapshot string
|
|
// Why is the hand act's words for a cancel (design 47 G A-117, §560).
|
|
Why string
|
|
}
|
|
|
|
// RecoveryWaitMinutes is the default of the manifest's config key `recovery-wait-minutes`, the bound of the wait for an
|
|
// ending event, counted from the first ask (design 47 G §554, §633).
|
|
const RecoveryWaitMinutes = 90
|
|
|
|
// Recovery is what the node agent needs of the recovery module and the manifests to move kept data. Nil moves none.
|
|
type Recovery struct {
|
|
// FormChanges are the kept data whose form differs between the running build's manifest and the applied build's (the
|
|
// catalog half's entries by fingerprint); an error says it is not known.
|
|
FormChanges func(ctx context.Context, module, runningFp, appliedFp string) ([]catalogue.FormChange, error)
|
|
// The seat's verbs, asked in the node-engine's own name; an error is the ask not answered.
|
|
SnapshotKept func(ctx context.Context, in meshrecovery.SnapshotKeptInput) (meshrecovery.SnapshotKeptAnswer, error)
|
|
RestoreKept func(ctx context.Context, in meshrecovery.RestoreKeptInput) (meshrecovery.RestoreKeptAnswer, error)
|
|
KeptPoints func(ctx context.Context, in meshrecovery.KeptPointsInput) (meshrecovery.KeptPointsAnswer, error)
|
|
// Hear follows the ending events of kind "ended" or "restored" of one module's kept data: the last one said already first,
|
|
// then each after it, until stop is called. It is made before the verb is asked (design 47 G §560).
|
|
Hear func(ctx context.Context, kind, module, kept string) (events <-chan Ending, stop func(), err error)
|
|
// Bound writes the bound key `<what>.<id>` of a wait in the state bounds with its time to live, so its end is the bus's
|
|
// marker, heard as expired (design 47 G A-77, A-92); end deletes it. Nil keeps no bound.
|
|
Bound func(ctx context.Context, what, id, module string, since time.Time, d time.Duration) (expired <-chan struct{}, end func(), err error)
|
|
// CatchUp calls the module's tool on this node through the tool runner (design 47 G A-108).
|
|
CatchUp func(ctx context.Context, module string, c meshrecovery.CatchUp) error
|
|
// WaitMinutes is the bound in minutes, the config key recovery-wait-minutes; 0 reads as RecoveryWaitMinutes.
|
|
WaitMinutes int
|
|
// Now is the clock; nil reads the machine's.
|
|
Now func() time.Time
|
|
}
|
|
|
|
func (r *Recovery) now() time.Time {
|
|
if r.Now != nil {
|
|
return r.Now()
|
|
}
|
|
return time.Now()
|
|
}
|
|
|
|
func (r *Recovery) bound() time.Duration {
|
|
if r.WaitMinutes > 0 {
|
|
return time.Duration(r.WaitMinutes) * time.Minute
|
|
}
|
|
return RecoveryWaitMinutes * time.Minute
|
|
}
|
|
|
|
// KeptMove is one kept data a want moves across a change of its form, as the seat mesh-recovery's verbs name it.
|
|
type KeptMove struct {
|
|
// Name is the kept data's local name, Form the form the restore point holds (the older one), and Fingerprint the
|
|
// fingerprint the restore point is taken before (the newer build's) or reverted from.
|
|
Name string
|
|
Form int
|
|
Fingerprint string
|
|
// CatchUp is, when taking the point, the tool the newer manifest declares beside the kept data.
|
|
CatchUp *meshrecovery.CatchUp
|
|
}
|
|
|
|
// keptMoves are the restore points a move needs: a point to take before the newer build starts (the kept data's form rises,
|
|
// or the newer manifest declares it no more) and a point to restore before the older build starts (the kept data goes back to
|
|
// a lower form). A kept data the running build does not declare is neither: nothing was written before it.
|
|
func keptMoves(changes []catalogue.FormChange, runningFp, appliedFp string) (take, restore []KeptMove) {
|
|
for _, c := range changes {
|
|
switch {
|
|
case c.Running == nil:
|
|
case c.Applied == nil || c.Applied.Form > c.Running.Form:
|
|
m := KeptMove{Name: c.Name, Form: c.Running.Form, Fingerprint: appliedFp}
|
|
if c.Applied != nil && c.Applied.CatchUp != nil {
|
|
m.CatchUp = &meshrecovery.CatchUp{Tool: c.Applied.CatchUp.Tool, Arguments: c.Applied.CatchUp.Arguments}
|
|
}
|
|
take = append(take, m)
|
|
case c.Applied.Form < c.Running.Form:
|
|
restore = append(restore, KeptMove{Name: c.Name, Form: c.Applied.Form, Fingerprint: runningFp})
|
|
}
|
|
}
|
|
return take, restore
|
|
}
|
|
|
|
// recoveryHeld is how a wait for the recovery module ended when the move cannot go on: the words are the reason of the hold.
|
|
type recoveryHeld struct {
|
|
Words string
|
|
Interrupted bool
|
|
}
|
|
|
|
// askAndWait asks the verb until its act is accepted and ended, as the table of design 47 G §560 says: `busy` waits for the
|
|
// ending of the act it names and asks again, `failed` with `stopped` asks again within the same bound, a failure with any
|
|
// other error holds. The wait is written in the have with its first ask's time and the act, so a restart waits for that act
|
|
// and asks nothing again; past the bound the have says the wait, and the wait goes on (the watchman raises it).
|
|
func (a *Agent) askAndWait(ctx context.Context, h Have, kind string, k KeptMove, ask func() (accepted, busy, errWords string, err error)) (Ending, *recoveryHeld) {
|
|
r := a.env.Recovery
|
|
module := h.Module
|
|
step := nodeengine.WaitStepCompose
|
|
wait := nodeengine.Wait{Owner: nodeengine.Seat, Step: step, On: "mesh-recovery." + kind + "." + module + "." + k.Name,
|
|
Reason: "recovery waits", Bound: r.bound().String()}
|
|
// A restart: the have names the wait with its ask time and act.
|
|
last, ok := a.env.Haves.Read(ctx, module)
|
|
if ok && last.Wait != nil && last.Wait.On == wait.On && last.Wait.AskAt != nil {
|
|
wait.AskAt, wait.Act, wait.Since = last.Wait.AskAt, last.Wait.Act, last.Wait.Since
|
|
}
|
|
now := r.now()
|
|
if wait.AskAt == nil {
|
|
wait.AskAt, wait.Since = &now, now
|
|
}
|
|
events, stop, err := r.Hear(ctx, kind, module, k.Name)
|
|
if err != nil {
|
|
return Ending{}, &recoveryHeld{Words: "the end of the recovery module's act cannot be heard: " + err.Error()}
|
|
}
|
|
defer stop()
|
|
var expired <-chan struct{}
|
|
if r.Bound != nil {
|
|
bound := r.bound() - now.Sub(*wait.AskAt)
|
|
if bound < time.Second {
|
|
bound = time.Second
|
|
}
|
|
exp, end, berr := r.Bound(ctx, "recovery", module+"."+kind+"."+k.Name, module, *wait.AskAt, bound)
|
|
if berr != nil {
|
|
return Ending{}, &recoveryHeld{Words: "the bound of the wait for the recovery module cannot be written: " + berr.Error()}
|
|
}
|
|
defer end()
|
|
expired = exp
|
|
}
|
|
write := func() {
|
|
w := wait
|
|
h.Wait = &w
|
|
a.writeHave(ctx, h)
|
|
}
|
|
act, busyOn := wait.Act, ""
|
|
for {
|
|
if act == "" && busyOn == "" {
|
|
write() // the first ask's time is in the have before the ask
|
|
accepted, busy, words, err := ask()
|
|
switch {
|
|
case err != nil:
|
|
if ctx.Err() != nil {
|
|
return Ending{}, &recoveryHeld{Words: BackupInterrupted, Interrupted: true}
|
|
}
|
|
return Ending{}, &recoveryHeld{Words: "the recovery module could not be asked: " + err.Error()}
|
|
case words != "":
|
|
return Ending{}, &recoveryHeld{Words: words}
|
|
case accepted != "":
|
|
act, wait.Act = accepted, accepted
|
|
write()
|
|
case busy != "":
|
|
busyOn = busy
|
|
default:
|
|
return Ending{}, &recoveryHeld{Words: "the recovery module answered neither accepted nor busy"}
|
|
}
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return Ending{}, &recoveryHeld{Words: BackupInterrupted, Interrupted: true}
|
|
case <-expired:
|
|
expired = nil
|
|
wait.Words = fmt.Sprintf("the recovery module has not ended its act within %s", r.bound())
|
|
write()
|
|
case ev, open := <-events:
|
|
if !open {
|
|
return Ending{}, &recoveryHeld{Words: "the end of the recovery module's act is no longer heard"}
|
|
}
|
|
switch {
|
|
case busyOn != "" && ev.Act == busyOn:
|
|
busyOn = ""
|
|
case act != "" && ev.Act == act:
|
|
switch {
|
|
case ev.Outcome == meshrecovery.EndedEventOutcomeDone:
|
|
return ev, nil
|
|
case ev.Error == "stopped":
|
|
act, wait.Act = "", ""
|
|
case ev.Error == "cancelled":
|
|
words := "cancelled"
|
|
if ev.Why != "" {
|
|
words += ": " + ev.Why
|
|
}
|
|
return ev, &recoveryHeld{Words: words}
|
|
default:
|
|
words := ev.Error
|
|
if words == "" {
|
|
words = "the recovery module's act failed"
|
|
}
|
|
return ev, &recoveryHeld{Words: words}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// keptStep takes the restore points a move of a module needs before it applies and restores the ones a revert needs: it
|
|
// answers whether the module is held (its have says why, and the running build is kept) and, after a restore, the catch-up
|
|
// of the point to call once the reverted build runs. A want that says to go on without a backup asks neither.
|
|
func (a *Agent) keptStep(ctx context.Context, module string, w wantNow, comp compNow, ranBefore bool, others []compositions.Composed) (held bool, catchUps []meshrecovery.CatchUp) {
|
|
r := a.env.Recovery
|
|
if !ranBefore || r == nil || r.FormChanges == nil || w.Want.Backup == meshwants.WantsByWantNodeModuleValueBackupSkip {
|
|
return false, nil
|
|
}
|
|
fp, dig := w.Want.Fp, w.Want.Composed
|
|
a.mu.Lock()
|
|
prev, runs := a.running[module]
|
|
a.mu.Unlock()
|
|
if !runs {
|
|
if h, found, _ := a.readHave(ctx, module); found {
|
|
prev = nodeengine.HaveByNodeModuleValue{Fp: h.FP, Composed: h.Composed}
|
|
}
|
|
}
|
|
if prev.Fp == "" || prev.Fp == fp {
|
|
return false, nil
|
|
}
|
|
changes, err := r.FormChanges(ctx, module, prev.Fp, fp)
|
|
if err != nil {
|
|
// Whether the move changes a form is not known: it waits for the entries and never moves without its restore point (design 47 G
|
|
// A-113). No attempt is kept, so the composer handing the composition again, or any settle, asks again.
|
|
words := fmt.Sprintf("whether %s's kept data changes form between %s and %s is not known, so it is not moved: %v", module, prev.Fp, fp, err)
|
|
a.sayOnce("kept:"+module+":"+fp, module+" waits: "+words)
|
|
since := r.now()
|
|
a.writeHave(ctx, Have{Module: module, FP: prev.Fp, Composed: prev.Composed, Phase: string(nodeengineclient.PhaseHeld), Detail: words,
|
|
Wait: &nodeengine.Wait{Owner: nodeengine.Seat, Step: nodeengine.WaitStepCompose, Reason: "entry-missing", Since: since, Words: words}})
|
|
if a.env.Waiting != nil {
|
|
a.env.Waiting(module, fp)
|
|
}
|
|
return true, nil
|
|
}
|
|
take, restore := keptMoves(changes, prev.Fp, fp)
|
|
if len(take)+len(restore) == 0 {
|
|
return false, nil
|
|
}
|
|
hold := func(words string, interrupted bool) {
|
|
if interrupted || ctx.Err() != nil {
|
|
words = BackupInterrupted
|
|
a.env.Say(fmt.Sprintf("%s waits: %s; it asks again at the next start", module, words))
|
|
} else {
|
|
a.env.Say(fmt.Sprintf("%s is held before %s: %s", module, fp, words))
|
|
}
|
|
a.mu.Lock()
|
|
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev, Held: true, AnsweredRev: w.Rev}
|
|
a.mu.Unlock()
|
|
since := r.now()
|
|
a.writeHave(ctx, Have{Module: module, FP: prev.Fp, Composed: prev.Composed, Phase: string(nodeengineclient.PhaseHeld), Detail: words,
|
|
Wait: &nodeengine.Wait{Owner: nodeengine.Seat, Step: nodeengine.WaitStepCompose, Reason: words, Since: since, Words: words}})
|
|
}
|
|
backing := Have{Module: module, FP: fp, Composed: dig, Phase: string(nodeengineclient.PhaseBackingUp)}
|
|
for _, m := range take {
|
|
why := fmt.Sprintf("form change of %s's %s to %s", module, m.Name, fp)
|
|
_, held := a.askAndWait(ctx, backing, "ended", m, func() (string, string, string, error) {
|
|
got, err := r.SnapshotKept(ctx, meshrecovery.SnapshotKeptInput{Module: module, Kept: m.Name, Form: int64(m.Form),
|
|
Fingerprint: m.Fingerprint, CatchUp: m.CatchUp, Why: why})
|
|
return got.Accepted, got.Busy, got.Error, err
|
|
})
|
|
if held != nil {
|
|
hold(fmt.Sprintf("the restore point of %s's %s (form %d) was not taken: %s", module, m.Name, m.Form, held.Words), held.Interrupted)
|
|
return true, nil
|
|
}
|
|
}
|
|
if len(restore) == 0 {
|
|
return false, nil
|
|
}
|
|
// A revert: the newer build stops first (it removes nothing), so the live data is not written while it is set aside.
|
|
stopped := false
|
|
if a.env.Stop != nil {
|
|
if res := a.env.Stop(ctx, module); res.Failed {
|
|
hold("the newer build could not be stopped before the restore: "+res.Detail, false)
|
|
return true, nil
|
|
}
|
|
stopped = true
|
|
}
|
|
// A revert that cannot restore starts the newer build again and holds the move, saying why: a module is never left stopped
|
|
// (design 47 G A-113).
|
|
holdRevert := func(words string, interrupted bool) {
|
|
if stopped {
|
|
words += "; " + a.startAgain(ctx, module, prev, others)
|
|
}
|
|
hold(words, interrupted)
|
|
}
|
|
for _, m := range restore {
|
|
pts, err := r.KeptPoints(ctx, meshrecovery.KeptPointsInput{Module: module, Kept: m.Name})
|
|
if err != nil {
|
|
holdRevert(fmt.Sprintf("restore point missing: the points of %s's %s cannot be read: %v", module, m.Name, err), false)
|
|
return true, nil
|
|
}
|
|
var point *meshrecovery.Point
|
|
for i := range pts.Points {
|
|
p := pts.Points[i]
|
|
if p.Module == module && p.Kept == m.Name && int(p.Form) == m.Form && p.Fingerprint == m.Fingerprint {
|
|
if point == nil || p.At.After(point.At) {
|
|
point = &pts.Points[i]
|
|
}
|
|
}
|
|
}
|
|
if point == nil {
|
|
holdRevert(fmt.Sprintf("restore point missing: %s's %s has none of form %d taken before %s", module, m.Name, m.Form, m.Fingerprint), false)
|
|
return true, nil
|
|
}
|
|
why := fmt.Sprintf("revert of %s's %s from %s", module, m.Name, m.Fingerprint)
|
|
_, held := a.askAndWait(ctx, backing, "restored", m, func() (string, string, string, error) {
|
|
got, err := r.RestoreKept(ctx, meshrecovery.RestoreKeptInput{Module: module, Kept: m.Name, Form: int64(m.Form),
|
|
Fingerprint: m.Fingerprint, Snapshot: point.Snapshot, Why: why})
|
|
return got.Accepted, got.Busy, got.Error, err
|
|
})
|
|
if held != nil {
|
|
holdRevert(fmt.Sprintf("%s's %s (form %d) is not restored: %s", module, m.Name, m.Form, held.Words), held.Interrupted)
|
|
return true, nil
|
|
}
|
|
if point.CatchUp != nil {
|
|
catchUps = append(catchUps, *point.CatchUp)
|
|
}
|
|
}
|
|
return false, catchUps
|
|
}
|
|
|
|
// startAgain starts the build a module ran before a revert stopped it, from the composition the agent applied or holds of its
|
|
// fingerprint, and says whether it did (design 47 G A-113).
|
|
func (a *Agent) startAgain(ctx context.Context, module string, prev nodeengine.HaveByNodeModuleValue, others []compositions.Composed) string {
|
|
a.mu.Lock()
|
|
c, ok := a.applied[module]
|
|
if !ok || c.C.Module == "" {
|
|
c, ok = a.compositionFor(module, prev.Fp)
|
|
}
|
|
a.mu.Unlock()
|
|
if !ok || a.env.Apply == nil {
|
|
return "the newer build could not be started again: its composition is not held"
|
|
}
|
|
res := a.env.Apply(context.WithoutCancel(ctx), c.C, others)
|
|
if res.Failed {
|
|
return "the newer build could not be started again: " + res.Detail
|
|
}
|
|
return "the newer build runs again"
|
|
}
|
|
|
|
// catchUp calls, once the reverted build of a module runs, the catch-up of each restore point restored, as the module's own
|
|
// tool through the tool runner (design 47 G A-108). A tool that fails is said: the build runs, and the tool can be called by hand.
|
|
func (a *Agent) catchUp(ctx context.Context, module string, calls []meshrecovery.CatchUp) {
|
|
if a.env.Recovery == nil || a.env.Recovery.CatchUp == nil {
|
|
return
|
|
}
|
|
for _, c := range calls {
|
|
if err := a.env.Recovery.CatchUp(ctx, module, c); err != nil {
|
|
a.env.Say(fmt.Sprintf("%s runs the reverted build, and its catch-up tool %s failed: %v", module, c.Tool, err))
|
|
continue
|
|
}
|
|
a.env.Say(fmt.Sprintf("%s's catch-up tool %s ran after the restore", module, c.Tool))
|
|
}
|
|
}
|