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",