diff --git a/cmd/mesh-host/keptartifact.go b/cmd/mesh-host/keptartifact.go new file mode 100644 index 00000000..0e7dcdb0 --- /dev/null +++ b/cmd/mesh-host/keptartifact.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode" + "github.com/novox/mesh-host/internal/apply" +) + +// The artifact store's holder takes its own new build from the build node that kept it (design 47): its done build names the +// node and the digest, the node's build seat answers where the bytes are as an object on the bus, and the node-engine +// reads them and checks them against the digest. Never another module's files: the seat's verb is the way. + +// keptAskWithin bounds one ask of a build seat; keptObjectWithin one read of the object. +var ( + keptAskWithin = 30 * time.Second + keptObjectWithin = 10 * time.Minute +) + +// keptBuildSeats are the three build seats, each of which serves the verb kept-artifact: the node that kept the build +// holds the seat of the language it was built in, and the others do not answer. +var keptBuildSeats = []string{nodebuildgo.Seat, nodebuildnode.Seat, nodebuildimage.Seat} + +// keptArtifactFetch is the apply's fetch of what a module's done build kept on a build node. +func keptArtifactFetch(buildOf func(context.Context, string, string) (string, string, bool, error), + ask func(ctx context.Context, node, seat, verb string, args any, within time.Duration) (json.RawMessage, error), + object func(ctx context.Context, store, name string, limit int64) ([]byte, error)) apply.KeptFetch { + return func(ctx context.Context, module, digest string) ([]byte, bool, error) { + node, fp, kept, err := buildOf(ctx, module, digest) + if err != nil || !kept { + return nil, false, err + } + hex := strings.TrimPrefix(digest, "sha256:") + var notHeld []string + for _, seat := range keptBuildSeats { + raw, err := ask(ctx, node, seat, nodebuildgo.VerbKeptArtifact, + nodebuildgo.KeptArtifactInput{Module: module, Fp: fp, Digest: hex}, keptAskWithin) + if err != nil { + notHeld = append(notHeld, seat+": "+err.Error()) + continue + } + var got nodebuildgo.KeptArtifactAnswer + if json.Unmarshal(raw, &got) != nil { + return nil, true, fmt.Errorf("%s of %s answered with something that is not an answer of kept-artifact", seat, node) + } + switch { + case got.Error != "": + return nil, true, fmt.Errorf("%s of %s could not give the kept build of %s: %s", seat, node, module, got.Error) + case got.Refused != "": + return nil, true, fmt.Errorf("%s of %s refuses the kept build of %s: %s %s", seat, node, module, got.Refused, got.Words) + case got.Store == "" || got.Object == "" || got.Digest != hex: + return nil, true, fmt.Errorf("%s of %s answered for %s without the store and the object of the digest %s", seat, node, module, hex) + } + octx, cancel := context.WithTimeout(ctx, keptObjectWithin) + body, err := object(octx, got.Store, got.Object, got.Size+1) + cancel() + if err != nil { + return nil, true, fmt.Errorf("the build of %s kept on %s could not be read from %s: %w", module, node, got.Store, err) + } + return body, true, nil + } + return nil, true, fmt.Errorf("%s keeps the build of %s for the artifact store and none of its build seats answers (%s)", + node, module, strings.Join(notHeld, "; ")) + } +} diff --git a/cmd/mesh-host/keptartifact_test.go b/cmd/mesh-host/keptartifact_test.go new file mode 100644 index 00000000..b662f01b --- /dev/null +++ b/cmd/mesh-host/keptartifact_test.go @@ -0,0 +1,73 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" +) + +// A module's done build that kept its artifact on a build node is fetched through that node's build seat (design 47). +func TestAKeptArtifactIsFetchedThroughTheBuildSeatOfItsNode(t *testing.T) { + hex := strings.Repeat("a", 64) + stored := func(string) (json.RawMessage, error) { + return json.RawMessage(`{"module":"distribution","fp":"f1","digest":"` + hex + `","size":5,"store":"kept-builds","object":"` + hex + `"}`), nil + } + for _, row := range []struct { + name string + kept bool + answer func(seat string) (json.RawMessage, error) + object error + wantKept bool + wantErr string + wantSeat string + }{ + {"a build kept nothing is not asked", false, nil, nil, false, "", ""}, + {"the seat of the language answers", true, func(seat string) (json.RawMessage, error) { + if seat != "node-build-node" { + return nil, errors.New("no responder") + } + return stored(seat) + }, nil, true, "", "node-build-node"}, + {"no build seat of the node answers", true, func(string) (json.RawMessage, error) { return nil, errors.New("no responder") }, + nil, true, "novox keeps the build of distribution", ""}, + {"the seat refuses it", true, func(string) (json.RawMessage, error) { + return json.RawMessage(`{"module":"distribution","fp":"f1","refused":"not kept","words":"swept"}`), nil + }, nil, true, "refuses the kept build of distribution: not kept", ""}, + {"the object cannot be read", true, stored, errors.New("no object"), true, "could not be read from kept-builds", "node-build-go"}, + } { + t.Run(row.name, func(t *testing.T) { + var seats []string + fetch := keptArtifactFetch( + func(context.Context, string, string) (string, string, bool, error) { return "novox", "f1", row.kept, nil }, + func(_ context.Context, node, seat, verb string, _ any, _ time.Duration) (json.RawMessage, error) { + if node != "novox" || verb != "kept-artifact" { + t.Errorf("asked %s of %s on %s", verb, seat, node) + } + seats = append(seats, seat) + return row.answer(seat) + }, + func(_ context.Context, store, name string, limit int64) ([]byte, error) { + if store != "kept-builds" || name != hex || limit < 5 { + t.Errorf("read %s/%s up to %d", store, name, limit) + } + return []byte("bytes"), row.object + }) + body, kept, err := fetch(context.Background(), "distribution", "sha256:"+hex) + if kept != row.wantKept { + t.Fatalf("kept = %v", kept) + } + if row.wantErr == "" && (err != nil || (kept && string(body) != "bytes")) { + t.Fatalf("got %q, %v", body, err) + } + if row.wantErr != "" && (err == nil || !strings.Contains(err.Error(), row.wantErr)) { + t.Fatalf("the error is %v, want %q", err, row.wantErr) + } + if !row.kept && len(seats) > 0 { + t.Errorf("a build that kept nothing asked %v", seats) + } + }) + } +} diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index ef053d56..de5b7c8d 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -28,6 +28,7 @@ import ( "github.com/novox/mesh-host/internal/seatwait" "github.com/novox/mesh-host/internal/selfcompose" "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" ) // The node agent's wiring (novox/hq ADR 0429 §4, ADR 0539 §9): the buckets it watches, the haves it writes, the backup @@ -89,7 +90,7 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool return agent.Have{}, false } return agent.Have{Module: module, FP: got.Fp, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, - Sealed: got.SealedAt}, true + Sealed: got.SealedAt, Wait: got.Wait}, true } // ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr). @@ -107,7 +108,7 @@ func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, b return agent.Have{}, false, err } return agent.Have{Module: module, FP: got.Fp, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, - Sealed: got.SealedAt}, true, nil + Sealed: got.SealedAt, Wait: got.Wait}, true, nil } func (h sessionHaves) Write(ctx context.Context, x agent.Have) { @@ -124,7 +125,7 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) { } h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase, Failed: phase == string(nodeengineclient.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits, - BuildAlone: x.BuildAlone, Kept: x.Kept, Sealed: x.Sealed, PullWait: x.PullWait}) + BuildAlone: x.BuildAlone, Kept: x.Kept, Sealed: x.Sealed, PullWait: x.PullWait, Wait: x.Wait}) } func (h sessionHaves) Removed(ctx context.Context, module string) { @@ -448,6 +449,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus } began := time.Now() node := session.Membership.Node + var keptFetch atomic.Pointer[apply.KeptFetch] // the composer's, set once it exists composedBucket, err := compositions.BucketFor(node) if err != nil { cancel() @@ -550,6 +552,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus Backoff: link.Backoff, Apply: func(ctx context.Context, c compositions.Composed, others []compositions.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) + if kf := keptFetch.Load(); kf != nil { + actx = apply.WithKept(actx, *kf) + } // While the grants part's list is followed, the controller's list in the bus module's composition is not // written: the apply keeps the list on the machine (ADR 0547 clause 8, never both). held, herr := users.Hold(c) @@ -574,6 +579,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus sayFacts(ctx, ag) return res }, + Stop: func(ctx context.Context, module string) agent.Result { + return stopComposition(ctx, opts, module) + }, Remove: func(ctx context.Context, module string, others []compositions.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) res := removeComposition(actx, opts, module, others, ag.Placed(module), sched, say) @@ -680,7 +688,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus go followRunning(run, session, composer, say) apply.UseEngineModule(composer.EngineModule) // the guard of the engine's own trees fixes no name (§527) env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday + env.Recovery = recoveryEnv(session, composer) env.WhyUnmade, env.Waiting = composer.WhyUnmade, composer.Waiting + keptFetchFn := keptArtifactFetch(composer.KeptBuildOf, session.AskSeatOn, session.KeptObject) + keptFetch.Store(&keptFetchFn) env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does session.NoteComposes(linked, nodeengineclient.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere") ag = agent.New(env) @@ -851,6 +862,30 @@ func removeComposition(ctx context.Context, opts options, module string, others return agent.Result{} } +// stopComposition stops a module's units and containers from the node's record and removes nothing (design 47 G A-98, +// §297): what the module kept stays, so a revert restores it before the reverted build starts. +func stopComposition(ctx context.Context, opts options, module string) agent.Result { + applying.Lock() + defer applying.Unlock() + unlock, err := store.Lock(opts.state, nil) + if err != nil { + return agent.Result{Failed: true, Detail: shorten(err.Error())} + } + defer unlock() + known, err := store.Load(opts.state) + if err != nil { + return agent.Result{Failed: true, Detail: shorten(err.Error())} + } + built, err := system.For(builtFor) + if err != nil { + return agent.Result{Failed: true, Detail: shorten(err.Error())} + } + if _, err := apply.StopModule(ctx, built, known, module, opts.run()); err != nil { + return agent.Result{Failed: true, Detail: shorten(err.Error())} + } + return agent.Result{} +} + // busyModules are the modules whose last apply left a container running because it said busy. func busyModules() []string { busyMu.Lock() diff --git a/cmd/mesh-host/recovery.go b/cmd/mesh-host/recovery.go new file mode 100644 index 00000000..9ce2acde --- /dev/null +++ b/cmd/mesh-host/recovery.go @@ -0,0 +1,89 @@ +package main + +import ( + "context" + "encoding/json" + "os" + "strconv" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/selfcompose" +) + +// recoveryEnv is what the node agent asks of the recovery module and the manifests to move a module's kept data (design 47 G §296, +// §297, §554, §560, §630, A-77, A-98, A-108): the seat's verbs asked in the node-engine's own name, the endings heard from the +// seat's events, the bound of a wait kept as a key of the state bounds, the catch-up called as the module's tool. +func recoveryEnv(session *link.Session, composer *selfcompose.Composer) *agent.Recovery { + ask := func(ctx context.Context, verb string, in, out any) error { + raw, err := session.AskMeshSeat(ctx, meshrecovery.Seat, verb, in) + if err != nil { + return err + } + return json.Unmarshal(raw, out) + } + return &agent.Recovery{ + FormChanges: composer.FormChanges, + SnapshotKept: func(ctx context.Context, in meshrecovery.SnapshotKeptInput) (a meshrecovery.SnapshotKeptAnswer, err error) { + return a, ask(ctx, meshrecovery.VerbSnapshotKept, in, &a) + }, + RestoreKept: func(ctx context.Context, in meshrecovery.RestoreKeptInput) (a meshrecovery.RestoreKeptAnswer, err error) { + return a, ask(ctx, meshrecovery.VerbRestoreKept, in, &a) + }, + KeptPoints: func(ctx context.Context, in meshrecovery.KeptPointsInput) (a meshrecovery.KeptPointsAnswer, err error) { + return a, ask(ctx, meshrecovery.VerbKeptPoints, in, &a) + }, + Hear: func(_ context.Context, kind, module, kept string) (<-chan agent.Ending, func(), error) { + heard, stop, err := session.HearRecoveryEnding(kind, module, kept) + if err != nil { + return nil, nil, err + } + out := make(chan agent.Ending, 64) + done := make(chan struct{}) + go func() { + for { + select { + case ev := <-heard: + select { + case out <- agent.Ending(ev): + case <-done: + return + } + case <-done: + return + } + } + }() + return out, func() { stop(); close(done) }, nil + }, + Bound: func(ctx context.Context, what, id, module string, since time.Time, d time.Duration) (<-chan struct{}, func(), error) { + return session.KeepBound(ctx, what, id, module, since, d) + }, + CatchUp: func(ctx context.Context, module string, c meshrecovery.CatchUp) error { + return session.CallTool(ctx, module, c.Tool, c.Arguments) + }, + WaitMinutes: recoveryWaitMinutes(), + } +} + +// recoveryWaitFile is the file the manifest declares with the content ${config:recovery-wait-minutes} (design 47 G §633, A-112): the +// node-engine is no bundle under a runtime, so it reads the setting from its file at its start, as the builder does (A-71). +var recoveryWaitFile = "/var/lib/mesh-host/recovery-wait-minutes" + +// recoveryWaitMinutes is the bound of the wait for the recovery module, in minutes: the file's number, the design's 90 when the +// file is missing, empty or not a positive number (design 47 G §554, A-112). +func recoveryWaitMinutes() int { + raw, err := os.ReadFile(recoveryWaitFile) + if err != nil { + return agent.RecoveryWaitMinutes + } + n, err := strconv.Atoi(strings.TrimSpace(string(raw))) + if err != nil || n <= 0 { + return agent.RecoveryWaitMinutes + } + return n +} diff --git a/cmd/mesh-host/recovery_integration_test.go b/cmd/mesh-host/recovery_integration_test.go new file mode 100644 index 00000000..2071af0d --- /dev/null +++ b/cmd/mesh-host/recovery_integration_test.go @@ -0,0 +1,142 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/nodegrant" +) + +// Design 47 G §560, §630, A-77, A-92, A-108: the node agent's recovery step over a real bus. Its session asks the seat's verbs, hears +// the last `restored` already said and the next, keeps a wait's bound as a key of the state bounds whose time ends with the bus's +// marker, and calls the catch-up as the module's tool with its arguments. The recovery module is a stand-in that answers the verbs and +// says the events; the grants part composes the node's subjects from the manifest, so the test adds them to the node's grant. +func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) { + perms := nodegrant.For(itNode) + perms.Publish = append(perms.Publish, "$JS.API.>", "$KV.mesh-host_bounds.>", "mesh.seat.mesh-recovery.call.*.node~"+itNode, + "mesh.mod.db.tool.rebuild."+itNode) + perms.Subscribe = append(perms.Subscribe, "mesh.seat.mesh-recovery.event.>", "$KV.mesh-host_bounds.>", "_INBOX.>", "$JS.ACK.>", "$JS.FC.>") + b := newItBedAs(t, perms) + ctx := context.Background() + + for _, s := range []struct{ stream, subject string }{{meshrecovery.StreamEnded, meshrecovery.SubjectEndedAll}, + {meshrecovery.StreamRestored, meshrecovery.SubjectRestoredAll}} { + if _, err := b.js.AddStream(&nats.StreamConfig{Name: s.stream, Subjects: []string{s.subject}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + } + js, err := jetstream.New(b.conn) + if err != nil { + t.Fatal(err) + } + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: link.BoundsBucket, History: 1, LimitMarkerTTL: time.Second}); err != nil { + t.Fatal(err) + } + say := func(act, outcome string) { + raw, _ := json.Marshal(meshrecovery.RestoredEvent{Act: act, Module: "db", Kept: "cache", Outcome: outcome, At: time.Now().UTC()}) + if _, err := b.js.Publish(meshrecovery.SubjectRestored("db", "cache"), raw); err != nil { + t.Fatal(err) + } + } + var restoreAsked meshrecovery.RestoreKeptInput + for verb, answer := range map[string]func(m *nats.Msg) any{ + meshrecovery.VerbKeptPoints: func(*nats.Msg) any { + return meshrecovery.KeptPointsAnswer{Points: []meshrecovery.Point{{Module: "db", Kept: "cache", Form: 1, Fingerprint: "f2", Snapshot: "snap-1", + CatchUp: &meshrecovery.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": true}}}}} + }, + meshrecovery.VerbRestoreKept: func(m *nats.Msg) any { + _ = json.Unmarshal(m.Data, &restoreAsked) + say("r1", meshrecovery.RestoredEventOutcomeDone) + return meshrecovery.RestoreKeptAnswer{Accepted: "r1"} + }, + } { + answer := answer + sub, err := b.conn.Subscribe("mesh.seat.mesh-recovery.call."+verb+".*", func(m *nats.Msg) { + out, _ := json.Marshal(map[string]any{"result": answer(m)}) + _ = m.Respond(out) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } + var toolArgs map[string]any + toolSub, err := b.conn.Subscribe(link.ToolSubject("db", "rebuild", itNode), func(m *nats.Msg) { + _ = json.Unmarshal(m.Data, &toolArgs) + _ = m.Respond([]byte(`{"result":{}}`)) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = toolSub.Unsubscribe() }) + + // A-112: the bound is the number in the file the manifest declares, 90 where it is missing or not a number. + for content, want := range map[string]int{"45\n": 45, "": 90, "soon\n": 90, "0\n": 90} { + recoveryWaitFile = filepath.Join(t.TempDir(), "w") + if err := os.WriteFile(recoveryWaitFile, []byte(content), 0o644); err != nil { + t.Fatal(err) + } + if got := recoveryWaitMinutes(); got != want { + t.Errorf("the file %q gives %d minutes, want %d", content, got, want) + } + } + recoveryWaitFile = filepath.Join(t.TempDir(), "missing") + if got := recoveryWaitMinutes(); got != 90 { + t.Errorf("no file gives %d minutes, want 90", got) + } + + say("old", meshrecovery.RestoredEventOutcomeFailed) // said before anyone listens: the last message of the subject + session := b.startSession() + r := recoveryEnv(session, nil) + + events, stop, err := r.Hear(ctx, "restored", "db", "cache") + if err != nil { + t.Fatal(err) + } + defer stop() + if first := <-events; first.Act != "old" || first.Outcome != "failed" { + t.Fatalf("the first event heard is %+v, want the last one said before", first) + } + pts, err := r.KeptPoints(ctx, meshrecovery.KeptPointsInput{Module: "db", Kept: "cache"}) + if err != nil || len(pts.Points) != 1 { + t.Fatalf("kept-points: %+v, %v", pts, err) + } + got, err := r.RestoreKept(ctx, meshrecovery.RestoreKeptInput{Module: "db", Kept: "cache", Form: 1, Fingerprint: "f2", Snapshot: pts.Points[0].Snapshot, Why: "w"}) + if err != nil || got.Accepted != "r1" || restoreAsked.Snapshot != "snap-1" { + t.Fatalf("restore-kept: %+v, %v, asked %+v", got, err, restoreAsked) + } + if ev := <-events; ev.Act != "r1" || ev.Outcome != "done" { + t.Fatalf("the ending heard is %+v", ev) + } + if err := r.CatchUp(ctx, "db", *pts.Points[0].CatchUp); err != nil || toolArgs["all"] != true { + t.Fatalf("catch-up: %v, arguments %v", err, toolArgs) + } + expired, end, err := r.Bound(ctx, "recovery", "db.restored.cache", "db", time.Now(), time.Second) + if err != nil { + t.Fatal(err) + } + defer end() + select { + case <-expired: + case <-time.After(10 * time.Second): + t.Fatal("the bound's end, the bus's marker, was not heard") + } +} + +// startSession is a session on the node's own connection, with no node agent run on it. +func (b *itBed) startSession() *link.Session { + b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second, + Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }} + b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode}) + return b.session +} diff --git a/cmd/mesh-host/stop_test.go b/cmd/mesh-host/stop_test.go new file mode 100644 index 00000000..4ca264a6 --- /dev/null +++ b/cmd/mesh-host/stop_test.go @@ -0,0 +1,40 @@ +package main + +import ( + "context" + "path/filepath" + "strings" + "testing" + + "github.com/novox/mesh-host/internal/store" +) + +// The node agent's Stop stops a module's container from the node's record, asks the machine nothing else and leaves the +// record, so the module's kept data and its record survive (design 47 G A-98). +func TestTheNodeAgentsStopStopsAModuleAndRemovesNothing(t *testing.T) { + was := builtFor + builtFor = "arch" + t.Cleanup(func() { builtFor = was }) + path := filepath.Join(t.TempDir(), "state.json") + var s store.State + s.Record(store.Applied{ID: "app.box", Type: "container", Target: "app-box"}) + s.Record(store.Applied{ID: "other.box", Type: "container", Target: "other-box"}) + if err := store.Save(path, s); err != nil { + t.Fatal(err) + } + var asked []string + opts := options{state: path, runner: func(_ context.Context, name string, args ...string) (string, error) { + asked = append(asked, name+" "+strings.Join(args, " ")) + return "", nil + }} + if res := stopComposition(context.Background(), opts, "app"); res.Failed { + t.Fatalf("stop failed: %s", res.Detail) + } + if !strings.Contains(strings.Join(asked, "\n"), "stop app-box") || strings.Contains(strings.Join(asked, "\n"), "other-box") || + strings.Contains(strings.Join(asked, "\n"), " rm ") { + t.Fatalf("the machine was asked %v", asked) + } + if after, err := store.Load(path); err != nil || len(after.Resources) != 2 { + t.Fatalf("the record changed: %v, %v", after.Resources, err) + } +} diff --git a/go.mod b/go.mod index 04fc37be..d40e3f1a 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,9 @@ require ( git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate v0.0.0-00010101000000-000000000000 @@ -30,6 +33,7 @@ require ( require ( git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/meshrecovery v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshvault v0.0.0-00010101000000-000000000000 diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 93def60b..ba620a77 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -75,6 +75,8 @@ type Have struct { // PullWait is, on a failed have, the pull from a store its apply failed on and waits on the holder of (issue 1966 // ENG-9, nodeengine.HaveByNodeModuleValue.PullWait): read through the have, as node_haves shows it. PullWait *nodeengine.PullWait + // Wait is the hold or wait the have says, in the recovery wait with its ask-at and act (design 47 G §560). + Wait *nodeengine.Wait } // Haves is the node's have bucket as the agent uses it. @@ -136,6 +138,13 @@ type Env struct { Apply func(ctx context.Context, c compositions.Composed, others []compositions.Composed) Result // Remove removes the module's own resources and nothing of another's. Remove func(ctx context.Context, module string, others []compositions.Composed) Result + // Stop stops the module's units and containers and removes nothing, so what it kept survives: a revert stops the newer + // build with it, restores the module's kept data, and then applies the reverted build (design 47 G A-98, §297). Nil + // stops nothing. + Stop func(ctx context.Context, module string) Result + // Recovery takes a module's kept data as a restore point before a form change moves it and restores it at a revert + // (design 47 G §296, §297, §554); nil moves none. + Recovery *Recovery // Backup backs a module up before a new composition of it is applied; nil backs up nothing. Backup func(ctx context.Context, module string) Backup // Kept is the resources the node last applied, by module, from the declaration it kept before the switch (ADR 0539 @@ -1628,6 +1637,12 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo return } + // A kept data on the bus whose form changes is taken as a restore point first, or restored at a revert (design 47 G §296, §297). + held, catchUps := a.keptStep(ctx, module, w, comp, ranBefore) + if held { + return + } + // A data module is backed up before a new composition of it moves (ADR 0429 §14), unless the want says to go on // without one (ADR 0539 §16). if comp.C.Data && ranBefore && a.env.Backup != nil && w.Want.Backup != meshwants.WantsByWantNodeModuleValueBackupSkip { @@ -1732,6 +1747,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo a.env.Settled(module) } a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(nodeengineclient.PhaseRunning), Detail: res.Detail}) + a.catchUp(ctx, module, catchUps) } // ranFailed marks what runs of a module as failed once an apply of it failed (novox/hq issue 1966, 2026-11-08): a failed diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 34268265..8a7b9791 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -930,6 +930,7 @@ func TestADataModuleIsBackedUpFirstAndHeldWhenItFails(t *testing.T) { // A data module with a good backup is applied after it; one that never ran has nothing to back up. func TestABackupDoneIsFollowedByTheApplyAndANewModuleIsNotBackedUp(t *testing.T) { + t.Run("a move across a kept data's form change is taken as a restore point and restored at a revert", testAMoveAcrossAFormChange) b := newBed(t) stop := b.run() defer stop() diff --git a/internal/agent/kept.go b/internal/agent/kept.go new file mode 100644 index 00000000..a4f38113 --- /dev/null +++ b/internal/agent/kept.go @@ -0,0 +1,323 @@ +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/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// 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 +} + +// 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 `.` 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": + return ev, &recoveryHeld{Words: "cancelled"} + 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) (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 { + // Not known whether the move changes a form: said, and the move goes on as before, since holding every module for a + // build record that is gone would stop deliveries a form never changed. + a.sayOnce("kept:"+module+":"+fp, fmt.Sprintf("%s: whether its kept data changes form between %s and %s is not known: %v", module, prev.Fp, fp, err)) + return false, 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. + 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 + } + } + for _, m := range restore { + pts, err := r.KeptPoints(ctx, meshrecovery.KeptPointsInput{Module: module, Kept: m.Name}) + if err != nil { + hold(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 { + hold(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 { + hold(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 +} + +// 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)) + } +} diff --git a/internal/agent/kept_test.go b/internal/agent/kept_test.go new file mode 100644 index 00000000..5c2f571b --- /dev/null +++ b/internal/agent/kept_test.go @@ -0,0 +1,257 @@ +package agent + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/novox/mesh-host/internal/compositions" +) + +// A recovery module that answers as scripted: each ask takes the next answer and says the events it carries after it. +type recoveryFake struct { + mu sync.Mutex + asks []string // verb and its input, in order + answers []fakeAnswer + points []meshrecovery.Point + events chan Ending + changes []catalogue.FormChange + calls []string // stop, ask, apply, catch-up, in order + catchUp []meshrecovery.CatchUp + expired chan struct{} +} + +type fakeAnswer struct { + accepted, busy, errWords string + then []Ending +} + +func (f *recoveryFake) next(verb string, in any) (string, string, string) { + f.mu.Lock() + defer f.mu.Unlock() + f.asks = append(f.asks, fmt.Sprintf("%s %s", verb, asJSON(in))) + f.calls = append(f.calls, verb) + if len(f.answers) == 0 { + return "", "", "no answer scripted" + } + a := f.answers[0] + f.answers = f.answers[1:] + for _, e := range a.then { + f.events <- e + } + return a.accepted, a.busy, a.errWords +} + +func (f *recoveryFake) env() *Recovery { + return &Recovery{ + FormChanges: func(context.Context, string, string, string) ([]catalogue.FormChange, error) { return f.changes, nil }, + SnapshotKept: func(_ context.Context, in meshrecovery.SnapshotKeptInput) (meshrecovery.SnapshotKeptAnswer, error) { + a, b, e := f.next("snapshot-kept", in) + return meshrecovery.SnapshotKeptAnswer{Accepted: a, Busy: b, Error: e}, nil + }, + RestoreKept: func(_ context.Context, in meshrecovery.RestoreKeptInput) (meshrecovery.RestoreKeptAnswer, error) { + a, b, e := f.next("restore-kept", in) + return meshrecovery.RestoreKeptAnswer{Accepted: a, Busy: b, Error: e}, nil + }, + KeptPoints: func(context.Context, meshrecovery.KeptPointsInput) (meshrecovery.KeptPointsAnswer, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "kept-points") + return meshrecovery.KeptPointsAnswer{Points: f.points}, nil + }, + Hear: func(context.Context, string, string, string) (<-chan Ending, func(), error) { + return f.events, func() {}, nil + }, + Bound: func(context.Context, string, string, string, time.Time, time.Duration) (<-chan struct{}, func(), error) { + return f.expired, func() {}, nil + }, + CatchUp: func(_ context.Context, _ string, c meshrecovery.CatchUp) error { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "catch-up") + f.catchUp = append(f.catchUp, c) + return nil + }, + Now: time.Now, + } +} + +func (f *recoveryFake) seen() []string { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.calls) +} + +// moveDB runs a data module at fp "a", then wants it at fp "b", and answers the bed once the second want is settled or held. +func moveDB(t *testing.T, f *recoveryFake, ranFrom, to string, settle func(b *bed) bool, before ...func(b *bed, dig string)) *bed { + t.Helper() + b := newBed(t) + inner, applies := b.env.Apply, 0 + b.env.Apply = func(ctx context.Context, c compositions.Composed, others []compositions.Composed) Result { + if applies++; applies >= 2 { + f.mu.Lock() + f.calls = append(f.calls, "apply") + f.mu.Unlock() + } + return inner(ctx, c, others) + } + f.events, f.expired = make(chan Ending, 16), make(chan struct{}) + b.env.Recovery = f.env() + b.env.Stop = func(context.Context, string) Result { + f.mu.Lock() + f.calls = append(f.calls, "stop") + f.mu.Unlock() + return Result{} + } + stop := b.run() + t.Cleanup(stop) + fp1, fp2 := fpOf(ranFrom), fpOf(to) + ck, craw, dig := composition(t, "db", fp1, true, file("db", "x", "1")) + wk, wraw := wantFor(t, "db", fp1, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "the first build is not applied", func() bool { return len(b.appliedList()) == 1 }) + b.idle() + ck2, craw2, dig2 := composition(t, "db", fp2, true, file("db", "x", "2")) + for _, h := range before { + h(b, dig) + } + wk2, wraw2 := wantFor(t, "db", fp2, dig2, "") + b.comps.put(ck2, craw2) + b.wants.put(wk2, wraw2) + eventually(t, "the move is neither applied nor held", func() bool { return settle(b) }) + return b +} + +// testAMoveAcrossAFormChange is design 47 G §296, §297, §554, §560, §630 to §632 and A-98, A-108 at the node agent. +func testAMoveAcrossAFormChange(t *testing.T) { + up := []catalogue.FormChange{{Name: "cache", Kind: catalogue.KeptKindState, Running: &catalogue.KeptLayout{Form: 1}, + Applied: &catalogue.KeptLayout{Form: 2, CatchUp: &catalogue.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": true}}}}} + down := []catalogue.FormChange{{Name: "cache", Kind: catalogue.KeptKindState, Running: &catalogue.KeptLayout{Form: 2}, + Applied: &catalogue.KeptLayout{Form: 1}}} + fp := func(c string) string { return fpOf(c) } + done := func(act string) Ending { return Ending{Act: act, Outcome: meshrecovery.EndedEventOutcomeDone} } + failed := func(act, why string) Ending { + return Ending{Act: act, Outcome: meshrecovery.EndedEventOutcomeFailed, Error: why} + } + applied := func(b *bed) bool { return len(b.appliedList()) == 2 } + heldHave := func(b *bed) bool { return b.haves.last("db").Phase == "held" } + + t.Run("a form that rises is taken as a restore point before the move, with the why and the catch-up, and then applied", func(t *testing.T) { + f := &recoveryFake{changes: up, answers: []fakeAnswer{{accepted: "a1", then: []Ending{done("a1")}}}} + moveDB(t, f, "a", "b", applied) + if got := f.seen(); !slices.Equal(got, []string{"snapshot-kept", "apply"}) { + t.Fatalf("calls %v", got) + } + want := fmt.Sprintf("snapshot-kept %s", asJSON(meshrecovery.SnapshotKeptInput{Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("b"), + Why: "form change of db's cache to " + fp("b"), CatchUp: &meshrecovery.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": true}}})) + if f.asks[0] != want { + t.Fatalf("asked %s, want %s", f.asks[0], want) + } + }) + + t.Run("busy waits for the other act's ending and asks again, and a stopped act is asked again; another act's failure wakes nothing", func(t *testing.T) { + f := &recoveryFake{changes: up, answers: []fakeAnswer{ + {busy: "x1", then: []Ending{failed("zz", "cap"), done("x1")}}, + {accepted: "a2", then: []Ending{failed("a2", "stopped")}}, + {accepted: "a3", then: []Ending{done("a3")}}, + }} + moveDB(t, f, "a", "b", applied) + if got := f.seen(); !slices.Equal(got, []string{"snapshot-kept", "snapshot-kept", "snapshot-kept", "apply"}) { + t.Fatalf("calls %v", got) + } + }) + + for _, row := range []struct{ error, held string }{ + {"cancelled", "cancelled"}, + {"cap", "cap"}, + {"restore point damaged", "restore point damaged"}, + {"disk full", "disk full"}, + } { + t.Run("an ending failed with "+row.error+" holds the move and asks nothing again", func(t *testing.T) { + f := &recoveryFake{changes: up, answers: []fakeAnswer{{accepted: "a1", then: []Ending{failed("a1", row.error)}}}} + b := moveDB(t, f, "a", "b", heldHave) + if h := b.haves.last("db"); h.FP != fp("a") || !contains(h.Detail, row.held) || h.Wait == nil || h.Wait.Reason == "" { + t.Fatalf("the have %+v", h) + } + if got := f.seen(); len(got) != 1 { + t.Fatalf("asked again: %v", got) + } + if n := len(b.appliedList()); n != 1 { + t.Fatalf("applied %d", n) + } + }) + } + + t.Run("the bound passed says the wait in the have and the wait goes on to the ending", func(t *testing.T) { + f := &recoveryFake{changes: up, answers: []fakeAnswer{{accepted: "a1"}}} + b := moveDB(t, f, "a", "b", func(b *bed) bool { + b.haves.mu.Lock() + defer b.haves.mu.Unlock() + for _, h := range b.haves.log { + if h.Wait != nil && h.Wait.Words != "" && h.Phase == "backing-up" { + return true + } + } + return false + }, func(*bed, string) { close(f.expired) }) + f.events <- done("a1") + eventually(t, "the move went on after the ending", func() bool { return applied(b) }) + }) + + t.Run("a revert stops the newer build, restores the point of the form it reverts from, applies, then calls its catch-up", func(t *testing.T) { + f := &recoveryFake{changes: down, + points: []meshrecovery.Point{ + {Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("z"), Snapshot: "other", At: time.Now()}, + {Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("a"), Snapshot: "snap-1", At: time.Now(), CatchUp: &meshrecovery.CatchUp{Tool: "rebuild"}}}, + answers: []fakeAnswer{{accepted: "r1", then: []Ending{done("r1")}}}} + moveDB(t, f, "a", "b", func(b *bed) bool { return len(f.seen()) >= 5 }) + if got := f.seen(); !slices.Equal(got, []string{"stop", "kept-points", "restore-kept", "apply", "catch-up"}) { + t.Fatalf("calls %v", got) + } + if f.catchUp[0].Tool != "rebuild" { + t.Fatalf("catch-up %+v", f.catchUp) + } + want := fmt.Sprintf("restore-kept %s", asJSON(meshrecovery.RestoreKeptInput{Module: "db", Kept: "cache", Form: 1, Fingerprint: fp("a"), + Snapshot: "snap-1", Why: "revert of db's cache from " + fp("a")})) + if f.asks[0] != want { + t.Fatalf("asked %s, want %s", f.asks[0], want) + } + }) + + t.Run("a revert with no point of its form holds as a restore point missing", func(t *testing.T) { + f := &recoveryFake{changes: down} + b := moveDB(t, f, "a", "b", heldHave) + if h := b.haves.last("db"); !contains(h.Detail, "restore point missing") { + t.Fatalf("the have %+v", h) + } + }) + + t.Run("a restart waits for the act its have names and asks nothing again", func(t *testing.T) { + f := &recoveryFake{changes: up} + b := moveDB(t, f, "a", "b", applied, func(b *bed, dig string) { + at := time.Now().Add(-time.Minute) + b.haves.mu.Lock() + b.haves.stored["db"] = Have{Module: "db", FP: fp("a"), Phase: "backing-up", Wait: &nodeengine.Wait{ + On: "mesh-recovery.ended.db.cache", AskAt: &at, Act: "a1", Since: at}} + b.haves.mu.Unlock() + f.events <- done("a1") + }) + _ = b + if got := f.seen(); !slices.Equal(got, []string{"apply"}) { + t.Fatalf("calls %v", got) + } + }) +} + +func contains(s, sub string) bool { return strings.Contains(s, sub) } + +func asJSON(v any) string { raw, _ := json.Marshal(v); return string(raw) } diff --git a/internal/apply/fetch_first.go b/internal/apply/fetch_first.go index ad91ec02..e4645f80 100644 --- a/internal/apply/fetch_first.go +++ b/internal/apply/fetch_first.go @@ -60,10 +60,21 @@ func fetchFirst(ctx context.Context, ordered []declaration.Resource, d *declarat continue } } - got, seen := bySource[source] - if !seen { + // What the module's done build kept on a build node is fetched from there, not from the store (design 47). + cacheKey, kept := source, false + if kf := keptFetchFrom(ctx); kf != nil && ours { + cacheKey = "kept " + module + " " + digest + if _, seen := bySource[cacheKey]; seen { + kept = true + } else if body, isKept, err := kf(ctx, module, digest); isKept { + kept = true + bySource[cacheKey] = fetchedArtifact{body: body, err: err} + } + } + got, seen := bySource[cacheKey] + if !seen && !kept { got.body, got.err = fetchMinding(ctx, source, in.away) - bySource[source] = got + bySource[cacheKey] = got } if got.err == nil { sum := sha256.Sum256(got.body) diff --git a/internal/apply/kept_artifact.go b/internal/apply/kept_artifact.go new file mode 100644 index 00000000..3585470a --- /dev/null +++ b/internal/apply/kept_artifact.go @@ -0,0 +1,29 @@ +package apply + +import "context" + +// The build the artifact store's holder keeps on a build node (novox/hq design 47, *The artifact store's holder takes its +// own new build even when its seat is the one broken*): the holder's program cannot be fetched from the store it serves, so +// its done build names the node that keeps it and the node-engine fetches it from there, through the build seat's verb. +// How is the caller's (a hook on the apply's context, as the files' link is); what apply does is use the bytes in place of +// the source's fetch, check them against the declared digest as every fetch is, and gate the module when they cannot be had. + +// KeptFetch gives the bytes of the artifact a module's done build kept on a build node, by the artifact's digest +// ("sha256:" and 64 hex digits, as a declaration names it). kept is false when the module's build kept no such artifact: +// the artifact is then fetched from its source, as before. An error says the artifact is kept and could not be had. +type KeptFetch func(ctx context.Context, module, digest string) (body []byte, kept bool, err error) + +type keptFetchKey struct{} + +// WithKept is ctx carrying how a kept artifact is fetched. +func WithKept(ctx context.Context, f KeptFetch) context.Context { + if f == nil { + return ctx + } + return context.WithValue(ctx, keptFetchKey{}, f) +} + +func keptFetchFrom(ctx context.Context) KeptFetch { + f, _ := ctx.Value(keptFetchKey{}).(KeptFetch) + return f +} diff --git a/internal/apply/replace_image_test.go b/internal/apply/replace_image_test.go index fa48c838..a7fd05d5 100644 --- a/internal/apply/replace_image_test.go +++ b/internal/apply/replace_image_test.go @@ -164,6 +164,47 @@ func TestARunningContainerIsKeptWhenItsReplacementImageCannotBeObtained(t *testi t.Errorf("the container's outcome is %+v, want skipped", o) } }) + + // Design 47 (W13.8): the holder's own program is fetched from the build node that kept it, not from the store; when it + // cannot be had, or is not the declared digest, nothing of the module is replaced and the failure says so. + for _, row := range []struct { + name string + fetch KeptFetch + wantErr string + }{ + {"the node that keeps it is away", func(context.Context, string, string) ([]byte, bool, error) { + return nil, true, errors.New("novox keeps it and does not answer") + }, "novox keeps it and does not answer"}, + {"what the node kept is not the declared digest", func(context.Context, string, string) ([]byte, bool, error) { + return []byte("other"), true, nil + }, "what arrived is"}, + } { + t.Run("a kept artifact: "+row.name, func(t *testing.T) { + gone := httptest.NewServer(http.NotFoundHandler()) + source := gone.URL + "/program.tar.gz" + gone.Close() + was := storeAwayAtMost + storeAwayAtMost = 0 + t.Cleanup(func() { storeAwayAtMost = was }) + d := parseTrusted(t, `{"declaration":1,"resources":[ + {"id":"distribution.program","type":"process","name":"program","source":"`+source+`", + "digest":"sha256:`+strings.Repeat("0", 64)+`","run":["program"]} + ]}`) + var asked []string + fetch := func(ctx context.Context, module, digest string) ([]byte, bool, error) { + asked = append(asked, module+" "+digest) + return row.fetch(ctx, module, digest) + } + run := func(ctx context.Context, name string, args ...string) (string, error) { return "", nil } + _, _, err := Apply(WithKept(context.Background(), fetch), archHost(t), d, store.State{}, store.OriginCarried, run, nil, nil) + if err == nil || !strings.Contains(err.Error(), row.wantErr) { + t.Errorf("the failure is %v, want it to say %q", err, row.wantErr) + } + if len(asked) != 1 || asked[0] != "distribution sha256:"+strings.Repeat("0", 64) { + t.Errorf("the kept fetch was asked %q", asked) + } + }) + } } func TestAReplacementImageIsPresentBeforeTheRunningContainerIsRemoved(t *testing.T) { diff --git a/internal/apply/stop.go b/internal/apply/stop.go new file mode 100644 index 00000000..8a8b553e --- /dev/null +++ b/internal/apply/stop.go @@ -0,0 +1,69 @@ +package apply + +import ( + "context" + "fmt" + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" +) + +// StopModule stops the units and containers the node's record holds for a module and removes nothing: no unit, file, +// container, data or record, so what the module kept survives and a later Apply starts the build it is given (design 47 +// G A-98, §297). Units come first, a timer before the service it starts, then containers; a unit the mesh never held +// (Stateless) is left as it is. It answers what it stopped, and the first failure. +func StopModule(ctx context.Context, sys system.System, known store.State, module string, run Runner) ([]string, error) { + var timers, services, containers []store.Applied + for _, r := range known.Resources { + if store.OwnerOf(r, declaration.ModuleGuess) != module { + continue + } + switch r.Type { + case string(declaration.TypeService): + if r.Stateless { + continue + } + if strings.HasSuffix(r.Target, ".timer") { + timers = append(timers, r) + } else { + services = append(services, r) + } + case string(declaration.TypeContainer): + containers = append(containers, r) + } + } + var stopped []string + for _, r := range append(timers, services...) { + manager := managerFor(r.Scope, r.User, run) + if state, err := sys.ServiceState(ctx, manager, r.Target); err != nil { + if strings.Contains(err.Error(), "does not exist on this machine") { + continue + } + return stopped, fmt.Errorf("stopping %s of %s: %w", r.Target, module, err) + } else if state == "stopped" { + continue + } + if err := sys.SetServiceState(ctx, manager, r.Target, "stopped"); err != nil { + return stopped, fmt.Errorf("stopping %s of %s: %w", r.Target, module, err) + } + stopped = append(stopped, r.Target) + } + if len(containers) == 0 { + return stopped, nil + } + cri, err := containerRuntime(ctx, run) + if err != nil { + return stopped, err + } + sort.Slice(containers, func(i, j int) bool { return containers[i].Target < containers[j].Target }) + for _, r := range containers { + if _, err := run(ctx, cri, "stop", r.Target); err != nil { + return stopped, fmt.Errorf("stopping the container %s of %s: %w", r.Target, module, err) + } + stopped = append(stopped, r.Target) + } + return stopped, nil +} diff --git a/internal/apply/stop_test.go b/internal/apply/stop_test.go new file mode 100644 index 00000000..2febfb5d --- /dev/null +++ b/internal/apply/stop_test.go @@ -0,0 +1,45 @@ +package apply + +import ( + "context" + "reflect" + "testing" + + "github.com/novox/mesh-host/internal/store" +) + +// Stop stops a module's timer, then its service, then its container, and leaves every record, and another module's +// units, as they are (design 47 G A-98). +func TestStopModuleStopsAModulesUnitsAndContainersAndRemovesNothing(t *testing.T) { + m := unitsMachine(map[string]*fakeUnit{ + "app.service": {active: "active", enabled: "enabled"}, + "app.timer": {active: "active", enabled: "enabled"}, + "other.service": {active: "active", enabled: "enabled"}, + }) + state := store.State{} + for _, r := range []store.Applied{ + {ID: "app.run", Type: "service", Target: "app.service"}, + {ID: "app.tick", Type: "service", Target: "app.timer"}, + {ID: "app.c", Type: "container", Target: "app-box"}, + {ID: "other.run", Type: "service", Target: "other.service"}, + {ID: "other.c", Type: "container", Target: "other-box"}, + } { + state.Record(r) + } + got, err := StopModule(context.Background(), archHost(t), state, "app", m.run) + if err != nil { + t.Fatal(err) + } + if want := []string{"app.timer", "app.service", "app-box"}; !reflect.DeepEqual(got, want) { + t.Fatalf("stopped %v, want %v, the timer first", got, want) + } + if m.units["app.service"].active != "inactive" || m.units["app.timer"].active != "inactive" || m.units["other.service"].active != "active" { + t.Fatalf("units are %+v", m.units) + } + if !m.did("docker stop app-box") || m.did("docker stop other-box") || m.did("docker rm") { + t.Fatalf("the machine was asked %v", m.asked) + } + if len(state.Resources) != 5 { + t.Fatalf("a record was removed: %d left", len(state.Resources)) + } +} diff --git a/internal/link/files.go b/internal/link/files.go index 44d02ddf..23f24eaa 100644 --- a/internal/link/files.go +++ b/internal/link/files.go @@ -172,6 +172,49 @@ func (l *natsLink) GrantsObject(ctx context.Context, name string, limit int64) ( return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.GrantsObject(ctx, name, limit) } +// KeptObject reads one object of a build module's store of kept builds by its name (design 47: the build the artifact store's +// holder keeps on a build node; the build seat's verb kept-artifact names the store and the object), with the same +// subjects as FileObject, verified against its digest. +func (b OverNATS) KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) { + js := b.JS + if b.Conn != nil { + wait := 10 * time.Second + if at, ok := ctx.Deadline(); ok { + if wait = time.Until(at); wait <= 0 { + return nil, context.DeadlineExceeded + } + } + bounded, err := b.Conn.JetStream(nats.MaxWait(wait)) + if err != nil { + return nil, err + } + js = bounded + } + return objectByName(ctx, js, store, declaration.FileStreamPrefix+store, name, limit) +} + +func (l *natsLink) KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) { + return OverNATS{Conn: l.conn, JS: l.js, Node: l.node}.KeptObject(ctx, store, name, limit) +} + +// KeptObjects is a link that can read an object of a build module's store of kept builds. +type KeptObjects interface { + KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) +} + +// KeptObject reads an object of a build module's store of kept builds on the link open now. +func (q *Session) KeptObject(ctx context.Context, store, name string, limit int64) ([]byte, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + k, ok := bus.(KeptObjects) + if bus == nil || !ok { + return nil, ErrNoLink + } + return k.KeptObject(ctx, store, name, limit) +} + // GrantsObjects is a link that can read an object of the grants part's bucket of sealed user lists. type GrantsObjects interface { GrantsObject(ctx context.Context, name string, limit int64) ([]byte, error) diff --git a/internal/link/have.go b/internal/link/have.go index ed4f2f36..e4c6bcf1 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -409,6 +409,7 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { if h.Phase == nodeengineclient.PhaseFailed { h.PullWait = f.PullWait } + h.Wait = f.Wait h.Health = s.healthOf(module, f) if h.Detail == "" { h.Detail = s.wordsOf(module, h.Health) diff --git a/internal/link/messages.go b/internal/link/messages.go index 16c32ebf..e84f31e6 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -647,4 +647,6 @@ type ModuleFact struct { Sealed compositions.Seals // PullWait is, on a failed fact, the pull from a store its apply waits on (nodeengine.HaveByNodeModuleValue.PullWait, issue 1966 ENG-9). PullWait *nodeengine.PullWait + // Wait is the hold or wait the module has, with the ask-at and act of a recovery wait (nodeengine.HaveByNodeModuleValue.Wait, design 47 G §560). + Wait *nodeengine.Wait } diff --git a/internal/link/recovery.go b/internal/link/recovery.go new file mode 100644 index 00000000..63be88bb --- /dev/null +++ b/internal/link/recovery.go @@ -0,0 +1,241 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// The node agent's wait for the recovery module (design 47 G §554, §560, A-77, A-92): the ending events of one module's kept data +// are heard through an ordered consumer that begins at the last message of the subject and follows the rest, and a wait's bound +// is a key of the state bounds with its own time to live, ended by the bus's marker. + +// BoundsBucket is the node-engine's state `bounds`, named by its holder as every state is. +const BoundsBucket = "mesh-host_bounds" + +// RecoveryBus is a link that hears the recovery module's events and keeps a bound. +type RecoveryBus interface { + // HearEnding calls heard with the last message of the subject on the stream, then each after it, until stop is called. + HearEnding(stream, subject string, heard func([]byte)) (stop func(), err error) + // BoundCreate writes the bound key with its time to live; one that stands is left as it is. BoundWatch calls ended once the + // bus's marker says the key's time ended; BoundDelete ends it. + BoundCreate(ctx context.Context, key string, value []byte, ttl time.Duration) error + BoundWatch(ctx context.Context, key string, ended func()) (stop func(), err error) + BoundDelete(ctx context.Context, key string) error +} + +func hearEnding(js nats.JetStreamContext, stream, subject string, heard func([]byte)) (func(), error) { + sub, err := js.Subscribe(subject, func(m *nats.Msg) { heard(m.Data) }, nats.OrderedConsumer(), nats.DeliverLast(), nats.BindStream(stream)) + if err != nil { + return nil, fmt.Errorf("the stream %s cannot be followed: %w", stream, err) + } + return func() { _ = sub.Unsubscribe() }, nil +} + +func boundKV(ctx context.Context, conn *nats.Conn) (jetstream.KeyValue, error) { + js, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := js.KeyValue(ctx, BoundsBucket) + if err != nil { + return nil, fmt.Errorf("the state %s cannot be bound: %w", BoundsBucket, err) + } + return kv, nil +} + +func boundCreate(ctx context.Context, conn *nats.Conn, key string, value []byte, ttl time.Duration) error { + kv, err := boundKV(ctx, conn) + if err != nil { + return err + } + if _, err := kv.Create(ctx, key, value, jetstream.KeyTTL(ttl)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) { + return err + } + return nil +} + +func boundWatch(ctx context.Context, conn *nats.Conn, key string, ended func()) (func(), error) { + kv, err := boundKV(ctx, conn) + if err != nil { + return nil, err + } + w, err := kv.Watch(ctx, key, jetstream.UpdatesOnly()) + if err != nil { + return nil, err + } + go func() { + for e := range w.Updates() { + if e != nil && e.Operation() == jetstream.KeyValuePurge { + ended() + } + } + }() + return func() { _ = w.Stop() }, nil +} + +func boundDelete(ctx context.Context, conn *nats.Conn, key string) error { + kv, err := boundKV(ctx, conn) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil +} + +func (l *natsLink) HearEnding(stream, subject string, heard func([]byte)) (func(), error) { + return hearEnding(l.js, stream, subject, heard) +} +func (l *natsLink) BoundCreate(ctx context.Context, key string, value []byte, ttl time.Duration) error { + return boundCreate(ctx, l.conn, key, value, ttl) +} +func (l *natsLink) BoundWatch(ctx context.Context, key string, ended func()) (func(), error) { + return boundWatch(ctx, l.conn, key, ended) +} +func (l *natsLink) BoundDelete(ctx context.Context, key string) error { + return boundDelete(ctx, l.conn, key) +} + +func (b OverNATS) HearEnding(stream, subject string, heard func([]byte)) (func(), error) { + return hearEnding(b.JS, stream, subject, heard) +} +func (b OverNATS) BoundCreate(ctx context.Context, key string, value []byte, ttl time.Duration) error { + return boundCreate(ctx, b.Conn, key, value, ttl) +} +func (b OverNATS) BoundWatch(ctx context.Context, key string, ended func()) (func(), error) { + return boundWatch(ctx, b.Conn, key, ended) +} +func (b OverNATS) BoundDelete(ctx context.Context, key string) error { + return boundDelete(ctx, b.Conn, key) +} + +var ( + _ RecoveryBus = (*natsLink)(nil) + _ RecoveryBus = OverNATS{} +) + +func (q *Session) recoveryBus() (RecoveryBus, error) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + h, ok := bus.(RecoveryBus) + if bus == nil || !ok { + return nil, ErrNoLink + } + return h, nil +} + +// RecoveryEnding is the ending event of an act as the node agent reads it, the fields every ending carries. +type RecoveryEnding struct { + Act string `json:"act"` + Outcome string `json:"outcome"` + Error string `json:"error,omitempty"` + Snapshot string `json:"snapshot,omitempty"` +} + +// HearRecoveryEnding follows the events `ended` (kind "ended") or `restored` (kind "restored") of one module's kept data: the last +// said first, then each after it, until stop is called. +func (q *Session) HearRecoveryEnding(kind, module, kept string) (<-chan RecoveryEnding, func(), error) { + bus, err := q.recoveryBus() + if err != nil { + return nil, nil, err + } + stream, subject := meshrecovery.StreamEnded, meshrecovery.SubjectEnded(module, kept) + if kind == "restored" { + stream, subject = meshrecovery.StreamRestored, meshrecovery.SubjectRestored(module, kept) + } + out := make(chan RecoveryEnding, 64) + stop, err := bus.HearEnding(stream, subject, func(raw []byte) { + var ev RecoveryEnding + if json.Unmarshal(raw, &ev) == nil && ev.Act != "" { + select { + case out <- ev: + default: + } + } + }) + if err != nil { + return nil, nil, err + } + return out, stop, nil +} + +// KeepBound writes the bound key `.` of a wait with its time to live and watches it: expired is closed when the bus's +// marker says its time ended, and end deletes it. +func (q *Session) KeepBound(ctx context.Context, what, id, module string, since time.Time, ttl time.Duration) (<-chan struct{}, func(), error) { + bus, err := q.recoveryBus() + if err != nil { + return nil, nil, err + } + key := what + "." + id + value, err := json.Marshal(struct { + Module string `json:"module,omitempty"` + Since string `json:"since"` + V int64 `json:"v"` + }{module, since.UTC().Format(time.RFC3339), 1}) + if err != nil { + return nil, nil, err + } + expired := make(chan struct{}) + stopWatch, err := bus.BoundWatch(ctx, key, func() { + select { + case <-expired: + default: + close(expired) + } + }) + if err != nil { + return nil, nil, err + } + if err := bus.BoundCreate(ctx, key, value, ttl); err != nil { + stopWatch() + return nil, nil, err + } + return expired, func() { + stopWatch() + _ = bus.BoundDelete(context.WithoutCancel(ctx), key) + }, nil +} + +// CallTool calls a module's tool on this machine's own node tools with its arguments and answers its refusal as an error (design 47 G +// A-108: the catch-up of a restore point is called so, as the node-engine is granted to). +func (q *Session) CallTool(ctx context.Context, module, tool string, args map[string]any) error { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + asker, ok := bus.(ToolBus) + if bus == nil || !ok { + return ErrNoLink + } + if args == nil { + args = map[string]any{} + } + body, err := json.Marshal(args) + if err != nil { + return err + } + reply, err := asker.Ask(ctx, ToolSubject(module, tool, q.Membership.Node), body) + if err != nil { + return err + } + var env struct { + Error string `json:"error"` + } + if err := json.Unmarshal(reply, &env); err != nil { + return fmt.Errorf("its answer is not a tool's answer: %w", err) + } + if env.Error != "" { + return errors.New(env.Error) + } + return nil +} diff --git a/internal/link/session.go b/internal/link/session.go index 2d74daf2..59bfb062 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -348,6 +348,12 @@ func SeatToolSubject(seat, verb, node string) string { // AskSeat asks the holder of a seat on this node a verb, over the link open now, and answers its result, or its refusal // as an error. The one call the node agent makes of another part: the backup before a data module moves (ADR 0429 §14). func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, within time.Duration) (json.RawMessage, error) { + return q.AskSeatOn(ctx, q.Membership.Node, seat, verb, args, within) +} + +// AskSeatOn asks the holder of a node seat on another node a verb: the build seat of the node a want's `kept-on` names +// gives the kept build (design 47 G section 526). +func (q *Session) AskSeatOn(ctx context.Context, node, seat, verb string, args any, within time.Duration) (json.RawMessage, error) { q.init() q.mu.Lock() bus := q.bus @@ -365,7 +371,7 @@ func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, with } asking, cancel := context.WithTimeout(ctx, within) defer cancel() - reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, q.Membership.Node), body) + reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, node), body) if err != nil { return nil, fmt.Errorf("%s.%s could not be asked: %w", seat, verb, err) } diff --git a/internal/selfcompose/entries.go b/internal/selfcompose/entries.go index 6b00c5b1..171d0092 100644 --- a/internal/selfcompose/entries.go +++ b/internal/selfcompose/entries.go @@ -15,6 +15,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/catalogue" "github.com/novox/mesh-host/internal/agent" ) @@ -514,3 +515,32 @@ func EntryCall(ask func(ctx context.Context, seat, verb string, args any) (json. // noEntryWords are the words the seat's holder refuses an unknown fingerprint in. const noEntryWords = "no entry of fingerprint" + +// FormChanges are the kept data whose form differs between the manifests of the build a module runs and the one a want moves it +// to, each read from the entry the composer holds of its fingerprint (design 47 G §296, §297). An entry not held is an error: the +// node agent then says it is not known and the move goes on. +func (c *Composer) FormChanges(_ context.Context, module, runningFp, appliedFp string) ([]catalogue.FormChange, error) { + if !c.byEntries() { + return nil, fmt.Errorf("this engine reads no entry by fingerprint") + } + read := func(fp string) (catalogue.Manifest, error) { + e, state := c.entryOf(fp) + if state != meshcatalogclient.EntryAnswered { + return catalogue.Manifest{}, fmt.Errorf("the entry of %s is %s", fp, state) + } + _, raw, err := entryManifest(e) + if err != nil { + return catalogue.Manifest{}, err + } + return catalogue.Decode(raw) + } + running, err := read(runningFp) + if err != nil { + return nil, fmt.Errorf("%s's running manifest: %w", module, err) + } + applied, err := read(appliedFp) + if err != nil { + return nil, fmt.Errorf("%s's wanted manifest: %w", module, err) + } + return catalogue.FormChanges(running, applied), nil +} diff --git a/internal/selfcompose/kept.go b/internal/selfcompose/kept.go new file mode 100644 index 00000000..52abee8c --- /dev/null +++ b/internal/selfcompose/kept.go @@ -0,0 +1,31 @@ +package selfcompose + +import ( + "context" + "strings" +) + +// KeptBuildOf is the node that keeps the build a module is wanted at, and its fingerprint, for the artifact store's +// holder (design 47 G section 271, 526, A-50): the want's `kept-on` and nothing else names it, the builder's builds being +// no read of the node-engine's (section 421). kept is false when the module has no want, its want names no keeping node, +// or its `parts` do not name the digest ("sha256:" and 64 hex digits, or bare): the artifact is then fetched from the store. +func (c *Composer) KeptBuildOf(_ context.Context, module, digest string) (node, fp string, kept bool, err error) { + c.mu.Lock() + w, ok := c.wants[module] + c.mu.Unlock() + if !ok || w.Fp == "" || w.KeptOn == "" { + return "", "", false, nil + } + if len(w.Parts) > 0 { + named := false + for _, d := range w.Parts { + if strings.TrimPrefix(d, "sha256:") == strings.TrimPrefix(digest, "sha256:") { + named = true + } + } + if !named { + return "", "", false, nil + } + } + return w.KeptOn, w.Fp, true, nil +} diff --git a/internal/selfcompose/kept_test.go b/internal/selfcompose/kept_test.go new file mode 100644 index 00000000..c2b7f305 --- /dev/null +++ b/internal/selfcompose/kept_test.go @@ -0,0 +1,35 @@ +package selfcompose + +import ( + "context" + "strings" + "testing" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" +) + +// The node that keeps a kept build is the want's kept-on alone, for an artifact its parts name (design 47 G 526, A-50). +func TestTheKeepingNodeIsTheWantsKeptOn(t *testing.T) { + kd := strings.Repeat("6", 64) + for _, row := range []struct { + name, digest string + want *meshwants.WantsByWantNodeModuleValue + node string + kept bool + }{ + {"kept, by its bare digest", kd, &meshwants.WantsByWantNodeModuleValue{Fp: "f1", KeptOn: "n3", Parts: map[string]string{"db:p": "sha256:" + kd}}, "n3", true}, + {"kept, by the declared digest", "sha256:" + kd, &meshwants.WantsByWantNodeModuleValue{Fp: "f1", KeptOn: "n3", Parts: map[string]string{"db:p": "sha256:" + kd}}, "n3", true}, + {"a digest the parts do not name", "sha256:" + strings.Repeat("7", 64), &meshwants.WantsByWantNodeModuleValue{Fp: "f1", KeptOn: "n3", Parts: map[string]string{"db:p": "sha256:" + kd}}, "", false}, + {"a want naming no keeping node", kd, &meshwants.WantsByWantNodeModuleValue{Fp: "f1", Parts: map[string]string{"db:p": "sha256:" + kd}}, "", false}, + {"no want of the module", kd, nil, "", false}, + } { + c := &Composer{wants: map[string]meshwants.WantsByWantNodeModuleValue{}} + if row.want != nil { + c.wants["db"] = *row.want + } + node, fp, kept, err := c.KeptBuildOf(context.Background(), "db", row.digest) + if err != nil || node != row.node || kept != row.kept || (kept && fp != "f1") { + t.Errorf("%s: node %q fp %q kept %v err %v", row.name, node, fp, kept, err) + } + } +} diff --git a/module.json b/module.json index f6f38d1f..29d7ff97 100644 --- a/module.json +++ b/module.json @@ -41,13 +41,18 @@ ], "invokes": [ "seat:mesh-catalog.entry", - "seat:mesh-catalog.seat-definitions" + "seat:mesh-catalog.seat-definitions", + "seat:mesh-recovery.kept-points", + "seat:mesh-recovery.restore-kept", + "seat:mesh-recovery.snapshot-kept" ], "consumes": [ "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-view.settled", + "mesh-recovery.ended", + "mesh-recovery.restored", "node-engine.running", "restic.backed-up" ], @@ -71,9 +76,20 @@ }, { "name": "packages" + }, + { + "name": "bounds", + "history": 1, + "key-ttl": true } ], "config": { + "recovery-wait-minutes": { + "type": "number", + "description": "How long the node agent waits for the recovery module to end an act it asked before a module moves across a change of a kept data's form, or at a revert, in minutes, counted from its first ask; past it the have says the wait (design 47 G section 554).", + "minimum": 1, + "default": 90 + }, "operator-signers": { "type": "string", "description": "The operator's public keys, ed25519 in base64, one a line: the node applies a composition made by hand only when one of these signed it (ADR 0539 clause 3). Public keys, no secret. Empty trusts no operator key.", @@ -108,6 +124,13 @@ "path": "/var/lib/mesh-host/operator-keys", "mode": "0644", "content": "${config:operator-signers}\n" + }, + { + "id": "recovery-wait", + "type": "file", + "path": "/var/lib/mesh-host/recovery-wait-minutes", + "mode": "0644", + "content": "${config:recovery-wait-minutes}\n" } ] } diff --git a/packaging/bounds_state_test.go b/packaging/bounds_state_test.go new file mode 100644 index 00000000..d92beab9 --- /dev/null +++ b/packaging/bounds_state_test.go @@ -0,0 +1,30 @@ +package packaging_test + +import ( + "os" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// The node-engine's waits run against bound keys with their own time to live, in its old section as its other states are +// (design 47 G A-92, §560, §627): the manifest declares the state bounds with a lifetime per key. +func TestTheManifestDeclaresTheBoundsStateWithAKeyLifetime(t *testing.T) { + raw, err := os.ReadFile("../module.json") + if err != nil { + t.Fatal(err) + } + m, err := catalogue.Decode(raw) + if err != nil { + t.Fatalf("the manifest no longer parses: %v", err) + } + for _, s := range m.State { + if s.Name == "bounds" { + if !s.KeyTTL || s.History != 1 { + t.Fatalf("the state bounds is %+v: one value a key, each key with its own lifetime", s) + } + return + } + } + t.Fatal("the manifest declares no state bounds") +} diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index d620eef7..fdf38d77 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -62,12 +62,28 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { } var m struct { Consumes []string `json:"consumes"` + Invokes []string `json:"invokes"` + Config map[string]struct { + Default any `json:"default"` + } `json:"config"` } if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } + // Design 47 G §630, §633, A-77: the recovery module's three verbs it asks, and its bound as a config key of 90. + for _, v := range []string{"seat:mesh-recovery.kept-points", "seat:mesh-recovery.restore-kept", "seat:mesh-recovery.snapshot-kept"} { + if !slices.Contains(m.Invokes, v) { + t.Errorf("invokes %v lacks %s", m.Invokes, v) + } + } + if !strings.Contains(string(raw), `"content": "${config:recovery-wait-minutes}\n"`) { + t.Errorf("module.json declares no file with the content ${config:recovery-wait-minutes} (design 47 G A-112)") + } + if m.Config["recovery-wait-minutes"].Default != float64(90) { + t.Errorf("config recovery-wait-minutes is %+v, want default 90", m.Config["recovery-wait-minutes"]) + } // The backup event, the status view's settled, and the two events of the seat that serves the seat list (ADR 0636 §4), the keeper's ask (design 47 G §387) and the seat's running (§199). - if want := []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-view.settled", "node-engine.running", "restic.backed-up"}; !slices.Equal(m.Consumes, want) { + if want := []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-view.settled", "mesh-recovery.ended", "mesh-recovery.restored", "node-engine.running", "restic.backed-up"}; !slices.Equal(m.Consumes, want) { t.Fatalf("consumes %v, want %v", m.Consumes, want) } // 774: the acts bucket mesh-host_acts is not declared as state for now: a declared state of a module with no account @@ -91,10 +107,11 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { // ADR 0565 clause 6: and the state packages, one key per node, its compositions' declared packages, read by pacman. // ADR 0564 §2, ADR 0636 §1: and the state findings, the node's finding that it waits for the seat list (no TTL: the // finding stays until the first full read deletes its key). - if len(parsed.State) != 4 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 || + // Design 47 G A-92: and the state bounds, each key a wait's bound with its own lifetime (packaging/bounds_state_test.go). + if len(parsed.State) != 5 || parsed.State[0].Name != "health" || parsed.State[0].TTLSeconds != 0 || parsed.State[1].Name != "findings" || parsed.State[1].TTLSeconds != 0 || parsed.State[2].Name != "data" || parsed.State[2].TTLSeconds != 0 || parsed.State[3].Name != "packages" || - parsed.State[3].TTLSeconds != 0 { + parsed.State[3].TTLSeconds != 0 || parsed.State[4].Name != "bounds" { t.Fatalf("the manifest declares the state %+v", parsed.State) } }