From 9e41f6380a1fc664fe2655d53f36ec594ee84125 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:37:55 +0100 Subject: [PATCH] mesh-recovery: the process, on its own account, answering the seat's verbs on the live bus MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The process reads its broker account, answers the seat mesh-recovery's verbs itself with the caller the bus proved, takes and restores streams through mesh-sdk's bussnap, keeps the digests in its state, names a kept data's bus objects from its catalog entry as mesh-sdk names them, learns the fingerprint a module runs at from the node-engine seat's running and the bus module's dumps from the backup holders' item events, each read whole at its start, and asks the backup holder on its node to bring the points directory back. A restore whose restore-started the bus did not acknowledge reads and sets aside nothing. Builds: design 47 G §554, §516, §551, §556, §563, §567, §655, A-107, A-112, A-121, A-123, A-126, A-132 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/cmd/mesh-recovery/bus.go | 98 ++++++++ .../mesh-recovery/cmd/mesh-recovery/dump.go | 163 +++++++++++++ .../mesh-recovery/cmd/mesh-recovery/main.go | 152 ++++++++++++ .../cmd/mesh-recovery/main_test.go | 68 ++++++ .../mesh-recovery/cmd/mesh-recovery/seats.go | 225 ++++++++++++++++++ modules/mesh-recovery/go.mod | 3 + .../internal/recovery/fakes_test.go | 4 + .../internal/recovery/restore.go | 7 +- .../internal/recovery/restore_test.go | 14 ++ modules/mesh-recovery/module.json | 3 + 10 files changed, 735 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/bus.go create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/dump.go create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/main.go create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/main_test.go create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/seats.go diff --git a/modules/mesh-recovery/cmd/mesh-recovery/bus.go b/modules/mesh-recovery/cmd/mesh-recovery/bus.go new file mode 100644 index 000000000..10dbac97c --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/bus.go @@ -0,0 +1,98 @@ +package main + +import ( + "context" + "errors" + "fmt" + "io" + "os" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-sdk/go/bussnap" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "mesh-recovery/internal/recovery" +) + +// liveBus is the bus as the recovery module's own account sees it (design 47 G §554): a stream's snapshot and restore +// through mesh-sdk's bussnap (§655, A-125), its deletion, and its durable consumers' positions. +type liveBus struct { + nc *nats.Conn + js jetstream.JetStream + scratch string + logf func(string, ...any) +} + +func (b liveBus) Objects(ctx context.Context, streams []string) ([]recovery.Object, error) { + var out []recovery.Object + for _, name := range streams { + s, err := b.js.Stream(ctx, name) + if errors.Is(err, jetstream.ErrStreamNotFound) { + out = append(out, recovery.Object{Stream: name}) + continue + } + if err != nil { + return nil, fmt.Errorf("stream %s: %w", name, err) + } + info, err := s.Info(ctx) + if err != nil { + return nil, fmt.Errorf("stream %s: %w", name, err) + } + o := recovery.Object{Stream: name, Exists: true, Bytes: int64(info.State.Bytes)} + names := s.ListConsumers(ctx) + for ci := range names.Info() { + if ci.Config.Durable == "" { + continue + } + o.Consumers = append(o.Consumers, seat.Consumer{Stream: name, Name: ci.Name, Sequence: int64(ci.AckFloor.Stream)}) + } + if err := names.Err(); err != nil { + return nil, fmt.Errorf("the consumers of %s: %w", name, err) + } + out = append(out, o) + } + return out, nil +} + +func (b liveBus) Take(ctx context.Context, stream string, w io.Writer) error { + _, err := bussnap.Snapshot(ctx, b.nc, w, bussnap.SnapshotOptions{Streams: []string{stream}, TempDir: b.scratch, Log: b.logf}) + return err +} + +func (b liveBus) Delete(ctx context.Context, stream string) error { + return bussnap.DeleteStream(ctx, b.nc, stream) +} + +func (b liveBus) Restore(ctx context.Context, file string) error { + scratch, err := os.MkdirTemp(b.scratch, ".restore-") + if err != nil { + return err + } + defer os.RemoveAll(scratch) + return bussnap.RestoreFile(ctx, b.nc, file, scratch, b.logf) +} + +// SetConsumers sets each durable consumer to its recorded sequence: the server's snapshot carries its consumers, so one +// restored with its stream stands as it was taken; one missing is made again from the sequence after its acknowledged one. +func (b liveBus) SetConsumers(ctx context.Context, cs []seat.Consumer) error { + for _, c := range cs { + s, err := b.js.Stream(ctx, c.Stream) + if err != nil { + return fmt.Errorf("stream %s: %w", c.Stream, err) + } + if _, err := s.Consumer(ctx, c.Name); err == nil { + continue + } else if !errors.Is(err, jetstream.ErrConsumerNotFound) { + return fmt.Errorf("the consumer %s of %s: %w", c.Name, c.Stream, err) + } + if _, err := s.CreateConsumer(ctx, jetstream.ConsumerConfig{Durable: c.Name, AckPolicy: jetstream.AckExplicitPolicy, + DeliverPolicy: jetstream.DeliverByStartSequencePolicy, OptStartSeq: uint64(c.Sequence) + 1}); err != nil { + return fmt.Errorf("the consumer %s of %s at %d: %w", c.Name, c.Stream, c.Sequence, err) + } + } + return nil +} + +var _ recovery.Bus = liveBus{} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/dump.go b/modules/mesh-recovery/cmd/mesh-recovery/dump.go new file mode 100644 index 000000000..3fb2d9b83 --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/dump.go @@ -0,0 +1,163 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/keys" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-recovery/internal/recovery" +) + +// The seat node-backup's item events (design 47 G §554, §563), each kept last per subject on a stream of its own. +const ( + backupSeat = "node-backup" + itemStarted = "item-started" + itemEnded = "item-ended" + itemStartedSteam = "SEAT_node-backup_item-started" + itemEndedStream = "SEAT_node-backup_item-ended" +) + +func itemSubject(event, node, module, item string) string { + return "mesh.seat." + backupSeat + ".event." + event + "." + node + "." + module + "." + item +} + +// dumpView is what the backup holders say of each item, `..` to the `at` of its last start and of its +// last end, read as §556 has it. A dump of an item runs while its last start is newer than its last end. +type dumpView struct { + mu sync.Mutex + started map[string]time.Time + ended map[string]time.Time + changed chan struct{} + // busModule is the module that holds the seat mesh-broker now (A-126), whose items are the bus's. + busModule func() (string, error) + js jetstream.JetStream +} + +func newDumpView(js jetstream.JetStream, busModule func() (string, error)) *dumpView { + return &dumpView{started: map[string]time.Time{}, ended: map[string]time.Time{}, changed: make(chan struct{}), + busModule: busModule, js: js} +} + +// hear takes one item event, an older one by `at` than the one held changing nothing, and wakes every wait. +func (d *dumpView) hear(m jetstream.Msg) { + d.take(m.Subject(), m.Data()) + _ = m.Ack() +} + +func (d *dumpView) take(subject string, data []byte) { + var body struct { + At time.Time `json:"at"` + } + rest, ok := strings.CutPrefix(subject, "mesh.seat."+backupSeat+".event.") + if !ok || json.Unmarshal(data, &body) != nil { + return + } + event, item, ok := strings.Cut(rest, ".") + if !ok { + return + } + d.mu.Lock() + defer d.mu.Unlock() + into := d.started + if event == itemEnded { + into = d.ended + } else if event != itemStarted { + return + } + if old, ok := into[item]; ok && body.At.Before(old) { + return + } + into[item] = body.At + close(d.changed) + d.changed = make(chan struct{}) +} + +// runs is the first item of the module whose dump runs, or "". +func (d *dumpView) runs(module string) string { + token := keys.EscapeName(module) + var on []string + for item, at := range d.started { + parts := strings.Split(item, ".") + if len(parts) != 3 || parts[1] != token { + continue + } + if end, ok := d.ended[item]; !ok || end.Before(at) { + on = append(on, item) + } + } + sort.Strings(on) + if len(on) == 0 { + return "" + } + return on[0] +} + +// readNow reads the last start and end of each of the bus module's items once more, through an ordered consumer made for +// that read (§556): the restore said its restore-started, acknowledged, before it reads (§563). +func (d *dumpView) readNow(ctx context.Context, module string) error { + for _, c := range []struct{ stream, event string }{{itemStartedSteam, itemStarted}, {itemEndedStream, itemEnded}} { + cons, err := d.js.OrderedConsumer(ctx, c.stream, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{itemSubject(c.event, "*", keys.EscapeName(module), "*")}, + DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return fmt.Errorf("the read of %s: %w", c.stream, err) + } + info, err := cons.Info(ctx) + if err != nil { + return fmt.Errorf("the read of %s: %w", c.stream, err) + } + for left := info.NumPending; left > 0; { + m, err := cons.Next(jetstream.FetchMaxWait(5 * time.Second)) + if err != nil { + return fmt.Errorf("the read of %s: %w", c.stream, err) + } + d.take(m.Subject(), m.Data()) + meta, err := m.Metadata() + if err != nil { + return err + } + left = meta.NumPending + } + } + return nil +} + +// Wait returns once no dump of the bus module's items runs; it calls waiting once, naming the item, when it has to wait +// (§554, §563). It wakes on the item events, never on a timer. +func (d *dumpView) Wait(ctx context.Context, waiting func(on string)) error { + module, err := d.busModule() + if err != nil { + return err + } + if err := d.readNow(ctx, module); err != nil { + return err + } + said := false + for { + d.mu.Lock() + on, ch := d.runs(module), d.changed + d.mu.Unlock() + if on == "" { + return nil + } + if !said { + waiting("the item-ended of the dump of " + on) + said = true + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ch: + } + } +} + +var _ recovery.Dump = (*dumpView)(nil) diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go new file mode 100644 index 000000000..91293e1fc --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -0,0 +1,152 @@ +// mesh-recovery: the recovery module's process (design 47 G §554). One process on an account of its own holds the seat +// mesh-recovery and answers its verbs itself over the bus with the caller the bus proved (A-107); it takes, sets aside and +// restores modules' kept data on the live bus, one act at a time, each act's result said as the seat's event. +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "os" + "os/signal" + "syscall" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatlist" + + "github.com/nats-io/nats.go" + + "mesh-recovery/internal/recovery" +) + +// The settings' files and the points directory, as the manifest's process section sets them (A-112, §633). +const ( + envPoints = "MESH_RECOVERY_POINTS" + envCap = "MESH_RECOVERY_POINTS_CAP_FILE" + envKeepDays = "MESH_RECOVERY_KEEP_DAYS_FILE" +) + +// The settings' defaults: the design's values (§551, §564). +const ( + defaultCap = 10 << 30 + defaultKeepDays = 90 +) + +// ExpireEvery is how often §551's removal runs beside its start and each snapshot-kept: once a day. +var ExpireEvery = 24 * time.Hour + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, os.Getenv, os.Stderr); err != nil { + fmt.Fprintln(os.Stderr, "mesh-recovery:", err) + os.Exit(1) + } +} + +// digests is the state digests, the points' digests kept on the bus apart from their bytes (§567). +type digests struct{ s process.State } + +func (d digests) Put(key string, v []byte) error { + _, err := d.s.Put(key, json.RawMessage(v)) + return err +} + +func (d digests) Get(key string) ([]byte, bool, error) { + e, err := d.s.Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil +} + +func (d digests) Delete(key string) error { return d.s.Delete(key) } + +// run is the process: it returns when ctx ends, after the running act ended. +func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { + logger := log.New(logOut, "mesh-recovery: ", log.LstdFlags) + dir := getenv(envPoints) + if dir == "" { + return errors.New(envPoints + " names no points directory, so nothing can be kept") + } + cap, err := readNumber(getenv(envCap), defaultCap) + if err != nil { + return err + } + keepDays, err := readNumber(getenv(envKeepDays), defaultKeepDays) + if err != nil { + return err + } + account, err := process.ReadBrokerAccount(getenv) + if err != nil { + return err + } + conn, err := process.Connect(account) + if err != nil { + return err + } + defer conn.Close() + nc, js, err := account.Dial(nats.Name(account.Node + "." + account.Module)) + if err != nil { + return err + } + defer nc.Close() + + // What the engine reads of the bus, each read as §556 has it before it acts on it: the node-engine seat's running + // (A-132) and the backup holders' item events of the bus module (§563). + running := newRunningView() + if _, err := account.HearSeatEvents(ctx, js, nodeengine.StreamRunning, running.hear, logger.Printf, process.OnRead(running.onRead)); err != nil { + return err + } + dump := newDumpView(js, func() (string, error) { return seatlist.HolderOf(conn.Call, meshbroker.Seat) }) + for _, s := range []string{itemStartedSteam, itemEndedStream} { + if _, err := account.HearSeatEvents(ctx, js, s, dump.hear, logger.Printf); err != nil { + return err + } + } + e := &recovery.Engine{ + Dir: dir, + Bus: liveBus{nc: nc, js: js, scratch: dir, logf: logger.Printf}, + Seats: liveSeats{emit: func(seat, event string, body any) error { return conn.EmitSeat(seat, event, body) }, + recorder: meshhandactsclient.Recorder{Ask: conn.Call}, call: conn.Call, callOn: conn.CallOn, + node: account.Node, points: dir, running: running}, + Digests: digests{conn.State("digests")}, + Dump: dump, + Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays)}, + Logf: logger.Printf, + } + for _, x := range e.Start(ctx) { + logger.Printf("the aside %s of %s's %s holds data the live bus lacks: recovery_restore_aside puts it back", x.Aside.ID, x.Module, x.Kept) + } + stopAnswers, err := conn.AnswerSeat(recovery.Seat, e.Handlers()) + if err != nil { + return fmt.Errorf("the seat %s's verbs cannot be answered: %w", recovery.Seat, err) + } + defer stopAnswers() + logger.Printf("started on the account %s, holding the seat %s", account.User, recovery.Seat) + expire := func(why string) { + if a := e.Accept(recovery.Act{Kind: recovery.KindExpire, Why: why, Caller: account.User}); a.Accepted == "" { + logger.Printf("the removal of points past their keep days is not run now (%s%s); it runs at the next snapshot-kept or day", a.Busy, a.Error) + } + } + expire("the removal of points past their keep days at the start (design 47 G section 551)") + tick := time.NewTicker(ExpireEvery) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + stopAnswers() + e.Wait() + return nil + case <-tick.C: + expire("the daily removal of points past their keep days (design 47 G section 551)") + } + } +} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go new file mode 100644 index 000000000..7f1d18dce --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// The process's own readings of the mesh (design 47 G §554, §563, A-121, A-132): the bus names of a kept data from its +// manifest, the fingerprint a module runs at from the node-engine seat's running, and whether a dump of the bus module runs. +func TestWhatTheProcessReads(t *testing.T) { + t.Run("bus names of a kept data (A-121)", func(t *testing.T) { + m, err := catalogue.Decode([]byte(`{"module":"app","version":"1","state":[{"name":"old"}],"logs":[{"name":"journal"}], + "streams":[{"name":"APP_EVENTS"}],"objects":[{"name":"files"}], + "requires":[{"name":"notes","storage":"key-value","template":"state"}]}`)) + if err != nil { + t.Fatal(err) + } + for kept, want := range map[string]string{"notes": "KV_app_notes OBJ_app_notes-objects", "old": "KV_app_old", + "journal": "LOG_app_journal", "APP_EVENTS": "APP_EVENTS", "files": "OBJ_app_files"} { + got, err := KeptStreams("app", kept, m) + if err != nil || strings.Join(got, " ") != want { + t.Errorf("%s: %v, %v; want %s", kept, got, err, want) + } + } + if _, err := KeptStreams("app", "none", m); err == nil { + t.Error("a kept data the manifest does not declare was named") + } + }) + t.Run("the running fingerprint (A-132)", func(t *testing.T) { + v := newRunningView() + if _, err := v.Fingerprint("app"); err == nil { + t.Error("answered before the read of running ended") + } + v.onRead(process.ReadDone) + at := time.Date(2026, 11, 12, 1, 0, 0, 0, time.UTC) + v.by["a\x00app"] = nodeengine.RunningEvent{Node: "a", Module: "app", Fingerprint: "src2:1", At: at} + v.by["b\x00app"] = nodeengine.RunningEvent{Node: "b", Module: "app", Stopped: true, At: at} + if fp, err := v.Fingerprint("app"); err != nil || fp != "src2:1" { + t.Errorf("%q, %v", fp, err) + } + v.by["b\x00app"] = nodeengine.RunningEvent{Node: "b", Module: "app", Fingerprint: "src2:2", At: at} + if _, err := v.Fingerprint("app"); err == nil || !strings.Contains(err.Error(), "src2:2 on b") { + t.Errorf("two fingerprints answered %v", err) + } + }) + t.Run("a dump of the bus module runs (§563)", func(t *testing.T) { + d := newDumpView(nil, nil) + start := time.Date(2026, 11, 12, 2, 0, 0, 0, time.UTC) + d.take(itemSubject(itemStarted, "srv", "nats", "data"), []byte(`{"at":"`+start.Format(time.RFC3339)+`"}`)) + d.take(itemSubject(itemStarted, "srv", "app", "data"), []byte(`{"at":"`+start.Format(time.RFC3339)+`"}`)) + if on := d.runs("nats"); on != "srv.nats.data" { + t.Errorf("runs %q", on) + } + d.take(itemSubject(itemEnded, "srv", "nats", "data"), []byte(`{"at":"`+start.Add(time.Minute).Format(time.RFC3339)+`","outcome":"done"}`)) + if on := d.runs("nats"); on != "" { + t.Errorf("an ended dump runs: %q", on) + } + d.take(itemSubject(itemEnded, "srv", "nats", "data"), []byte(`{"at":"`+start.Add(-time.Hour).Format(time.RFC3339)+`","outcome":"done"}`)) + if on := d.runs("nats"); on != "" { + t.Errorf("an older end changed what runs: %q", on) + } + }) +} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/seats.go b/modules/mesh-recovery/cmd/mesh-recovery/seats.go new file mode 100644 index 000000000..58937e038 --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/seats.go @@ -0,0 +1,225 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" + + "github.com/nats-io/nats.go/jetstream" + + "mesh-recovery/internal/recovery" +) + +// Call is how the process calls a seat's verb (process.Conn.Call), and CallOn one machine's holder of a node seat. +type ( + Call func(key string, body any) (json.RawMessage, error) + CallOn func(node, key string, body any) (json.RawMessage, error) +) + +// liveSeats is what the engine asks of the rest of the mesh, each by its seat (design 47 G §554). +type liveSeats struct { + emit func(seat, event string, body any) error + recorder meshhandactsclient.Recorder + call Call + callOn CallOn + node string + points string + running *runningView +} + +// Say says one event of the seat mesh-recovery, each part filled in as one token, escaped as a key's name is (§523). +func (s liveSeats) Say(event string, body any, parts ...string) error { + subject := event + for _, p := range parts { + if p == "" || p == "." || p == ".." { + return fmt.Errorf("the event %s cannot be said with the part %q", event, p) + } + subject += "." + keys.EscapeName(p) + } + return s.emit(recovery.Seat, subject, body) +} + +// HandAct writes the act in the seat mesh-hand-acts' log first, naming the account the bus proved and who asked (§554, §682). +func (s liveSeats) HandAct(caller, asker, what, why string) error { + line := what + " (the account " + orNone(caller) + if asker != "" { + line += ", asked by " + asker + } + _, err := s.recorder.Record(meshhandactsclient.Request{What: line + ")", Why: why, Verb: recovery.Seat}) + return err +} + +func orNone(s string) string { + if s == "" { + return "unproven" + } + return s +} + +// Streams are the bus names of a module's kept data at a fingerprint, from the catalog graph entry `mesh-catalog`'s `entry` +// answers (§554 Bus names, §443): a template entry's as mesh-sdk names them, an old section's as the bus composer of +// mesh-sdk names it when it makes it (A-121). +func (s liveSeats) Streams(_ context.Context, module, kept, fingerprint string) ([]string, error) { + raw, err := s.call("seat:"+meshcatalog.Seat+"."+meshcatalog.VerbEntry, meshcatalog.EntryInput{Fingerprint: fingerprint}) + if err != nil { + return nil, fmt.Errorf("the catalog entry of %s: %w", fingerprint, err) + } + var ans meshcatalog.EntryAnswer + if err := json.Unmarshal(raw, &ans); err != nil { + return nil, fmt.Errorf("the catalog entry of %s does not read: %w", fingerprint, err) + } + if ans.Entry.Module != module { + return nil, fmt.Errorf("the catalog entry of %s is %s's, not %s's", fingerprint, ans.Entry.Module, module) + } + whole, err := json.Marshal(ans.Entry.Manifest) + if err != nil { + return nil, err + } + m, err := catalogue.Decode(whole) + if err != nil { + return nil, fmt.Errorf("the manifest of %s at %s: %w", module, fingerprint, err) + } + return KeptStreams(module, kept, m) +} + +// KeptStreams names the bus objects of one kept data a manifest declares (A-121). +func KeptStreams(module, kept string, m catalogue.Manifest) ([]string, error) { + for _, e := range m.BusState { + if e.Name == kept { + if names := catalogue.BusStateStreams(module, e); len(names) > 0 { + return names, nil + } + return nil, fmt.Errorf("%s's %s has no bus name", module, kept) + } + } + for _, d := range catalogue.KeptDeclarations(m) { + if d.Name != kept { + continue + } + switch d.Kind { + case catalogue.KeptKindState: + return []string{"KV_" + buscompose.BucketName(module, kept)}, nil + case catalogue.KeptKindLog: + return []string{buscompose.LogStreamName(module, kept)}, nil + case catalogue.KeptKindStream: + return []string{kept}, nil + case catalogue.KeptKindObjects: + return []string{"OBJ_" + buscompose.BucketName(module, kept)}, nil + } + } + return nil, fmt.Errorf("%s's manifest declares no kept data %s", module, kept) +} + +// RunningFingerprint is the fingerprint the node-engine runs the module at, from the node-engine seat's running (A-123, A-132). +func (s liveSeats) RunningFingerprint(_ context.Context, module string) (string, error) { + return s.running.Fingerprint(module) +} + +// Backup brings the points directory back beside its path through the seat node-backup's restore on this node (§516), +// answering the directory it was brought back into. +func (s liveSeats) Backup(_ context.Context, module string) (string, error) { + if _, err := s.callOn(s.node, "seat:node-backup.restore", map[string]string{"module": module, "path": s.points}); err != nil { + return "", fmt.Errorf("the backup holder on %s brought nothing back: %w", s.node, err) + } + back, _ := filepath.Glob(s.points + ".restored-*") + if len(back) == 0 { + return "", fmt.Errorf("the backup holder on %s answered, and nothing lies beside %s", s.node, s.points) + } + sort.Strings(back) + return back[len(back)-1], nil +} + +var _ recovery.Seats = liveSeats{} + +// runningView is what the node-engine seat's running says of each node's module, read as §556 has it: the read of the last +// of each subject first, the durable followed after it, an older one by `at` changing nothing. +type runningView struct { + mu sync.Mutex + read bool + by map[string]nodeengine.RunningEvent +} + +func newRunningView() *runningView { return &runningView{by: map[string]nodeengine.RunningEvent{}} } + +func (v *runningView) hear(m jetstream.Msg) { + var r nodeengine.RunningEvent + if err := json.Unmarshal(m.Data(), &r); err == nil && r.Module != "" { + k := r.Node + "\x00" + r.Module + v.mu.Lock() + if old, ok := v.by[k]; !ok || !r.At.Before(old.At) { + v.by[k] = r + } + v.mu.Unlock() + } + _ = m.Ack() +} + +func (v *runningView) onRead(state string) { + v.mu.Lock() + v.read = state == process.ReadDone + v.mu.Unlock() +} + +// Fingerprint is the one fingerprint the module runs at on the nodes that run it; two different ones are refused, +// naming both, so a set-aside never takes the names of a build that does not run. +func (v *runningView) Fingerprint(module string) (string, error) { + v.mu.Lock() + defer v.mu.Unlock() + if !v.read { + return "", errors.New("the node-engine seat's running is still being read") + } + fps := map[string][]string{} + for _, r := range v.by { + if r.Module == module && !r.Stopped && r.Fingerprint != "" { + fps[r.Fingerprint] = append(fps[r.Fingerprint], r.Node) + } + } + switch len(fps) { + case 0: + return "", fmt.Errorf("no node runs %s", module) + case 1: + for fp := range fps { + return fp, nil + } + } + var said []string + for fp, nodes := range fps { + sort.Strings(nodes) + said = append(said, fp+" on "+strings.Join(nodes, ", ")) + } + sort.Strings(said) + return "", fmt.Errorf("%s runs at more than one fingerprint (%s)", module, strings.Join(said, "; ")) +} + +// readNumber reads a setting's file (design 47 G A-112): its whole content a positive number, the default when missing. +func readNumber(path string, fallback int64) (int64, error) { + if path == "" { + return fallback, nil + } + raw, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return fallback, nil + } + if err != nil { + return 0, err + } + var n float64 + if err := json.Unmarshal([]byte(strings.TrimSpace(string(raw))), &n); err != nil || n <= 0 { + return 0, fmt.Errorf("the setting in %s is %q, not a positive number", path, strings.TrimSpace(string(raw))) + } + return int64(n), nil +} diff --git a/modules/mesh-recovery/go.mod b/modules/mesh-recovery/go.mod index ee3bd0cc0..8056787cb 100644 --- a/modules/mesh-recovery/go.mod +++ b/modules/mesh-recovery/go.mod @@ -5,6 +5,9 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/go/seats/meshbroker 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/go/seats/nodeengine v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient 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-sdk/go v0.0.0-00010101000000-000000000000 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-recovery/internal/recovery/fakes_test.go b/modules/mesh-recovery/internal/recovery/fakes_test.go index 48796d2cb..69c471955 100644 --- a/modules/mesh-recovery/internal/recovery/fakes_test.go +++ b/modules/mesh-recovery/internal/recovery/fakes_test.go @@ -99,11 +99,15 @@ type fakeSeats struct { handFail error streams []string backup string + sayFail map[string]error } func (s *fakeSeats) Say(event string, body any, parts ...string) error { s.mu.Lock() defer s.mu.Unlock() + if err := s.sayFail[event]; err != nil { + return err + } s.events = append(s.events, said{event, body, parts}) return nil } diff --git a/modules/mesh-recovery/internal/recovery/restore.go b/modules/mesh-recovery/internal/recovery/restore.go index 744a2eaf6..fb6803b55 100644 --- a/modules/mesh-recovery/internal/recovery/restore.go +++ b/modules/mesh-recovery/internal/recovery/restore.go @@ -75,8 +75,11 @@ func (e *Engine) cancelledWhy(err error) string { // against the bus, sets the live data aside whole and checked, removes it, restores the point under its names and // sets each consumer to its recorded sequence; a step after the removal that fails puts the aside back. func (e *Engine) restore(ctx context.Context, a *Act) error { - e.say(seat.EventRestoreStarted, seat.RestoreStartedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, - Fingerprint: a.Fingerprint, Snapshot: a.Snapshot}, a.Module, a.Kept) + // Said means acknowledged: the dump's item events are read only once the bus took restore-started (§563). + if err := e.Seats.Say(seat.EventRestoreStarted, seat.RestoreStartedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, + Fingerprint: a.Fingerprint, Snapshot: a.Snapshot}, a.Module, a.Kept); err != nil { + return e.endedRestored(a, fmt.Errorf("restore-started was not acknowledged by the bus, so nothing was read or set aside: %w", err), nil, false) + } if e.Dump != nil { if err := e.Dump.Wait(ctx, func(on string) { e.setStep(a, "waiting", on) }); err != nil { if ctx.Err() != nil { diff --git a/modules/mesh-recovery/internal/recovery/restore_test.go b/modules/mesh-recovery/internal/recovery/restore_test.go index 853a68645..de2cbe0ff 100644 --- a/modules/mesh-recovery/internal/recovery/restore_test.go +++ b/modules/mesh-recovery/internal/recovery/restore_test.go @@ -259,3 +259,17 @@ func TestAsideMovesTheLiveDataIntoAnAsideAndRestoreAsidePutsItBack(t *testing.T) t.Errorf("streams %v", r.b.streams) } } + +// Said means acknowledged: a restore whose restore-started the bus did not take reads nothing and sets nothing aside, and +// ends failed (§563). +func TestARestoreWhoseStartIsNotAcknowledgedTouchesNothing(t *testing.T) { + r := takenRig(t) + r.s.sayFail = map[string]error{seat.EventRestoreStarted: errTest} + r.do(restoreAct("app/notes/2-fp1")) + if ev := lastRestored(t, r); ev.Outcome != "failed" || !strings.Contains(ev.Error, "restore-started") || r.b.streams["app_notes"] != "hello" { + t.Errorf("restored %+v, streams %v", ev, r.b.streams) + } + if kp, _ := r.e.KeptPoints("", ""); len(kp.Asides) != 0 { + t.Errorf("an aside was taken: %+v", kp.Asides) + } +} diff --git a/modules/mesh-recovery/module.json b/modules/mesh-recovery/module.json index ac477062c..5f502377f 100644 --- a/modules/mesh-recovery/module.json +++ b/modules/mesh-recovery/module.json @@ -35,11 +35,13 @@ "consumes": [ "node-backup.item-started", "node-backup.item-ended", + "node-engine.running", "mesh-broker.snapshot-restored", "mesh-conditions.findings-asked" ], "invokes": [ "seat:mesh-catalog.entry", + "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", "seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", @@ -223,6 +225,7 @@ ], "invokes": [ "seat:mesh-catalog.entry", + "seat:mesh-module-assigner.assignments", "seat:mesh-hand-acts.hand-act", "seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore",