From 87ca34c92f40ad0dc03ec9cde7b9f8436e4ac9f1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 16:52:20 +0100 Subject: [PATCH] A-291: the recovery keeper receives snapshot-get's chunks through the SDK's one helper The keeper's hand-made receive (subscribe to the call's chunks, ask, receive in order, acknowledge each) moves into mesh-sdk's process.Conn.CallChunks; the keeper's answer checks (an answered error, the space it fits, no chunk granted) run as its accept hook before a chunk is taken. Builds: design 47 G A-291 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/cmd/mesh-recovery/keeper.go | 76 +++++++++---------- .../mesh-recovery/cmd/mesh-recovery/main.go | 2 +- 2 files changed, 36 insertions(+), 42 deletions(-) diff --git a/modules/mesh-recovery/cmd/mesh-recovery/keeper.go b/modules/mesh-recovery/cmd/mesh-recovery/keeper.go index 1d60219c4..df395d37e 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/keeper.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/keeper.go @@ -5,15 +5,13 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "sync" - "time" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" - "git.novox.be/novox/mesh-sdk/go/buscalls" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "mesh-recovery/internal/recovery" @@ -22,11 +20,14 @@ import ( // liveKeeper is the bus module's keeper reached by the seat mesh-broker's verbs (design 47 G §681): snapshot-get's chunks // received in mesh-sdk's form (A-139), snapshot-restore and its cancel called, and snapshot-restored heard. type liveKeeper struct { - nc *nats.Conn call Call + chunks Chunks restored *restoredView } +// Chunks calls a seat verb whose answer is sent in chunks and receives them into w (process.Conn.CallChunks, design 47 G A-291). +type Chunks func(ctx context.Context, key, call string, body any, accept func(answer json.RawMessage) error, w io.Writer) (json.RawMessage, error) + func (k liveKeeper) verb(verb string, in, out any) error { raw, err := k.call("seat:"+meshbroker.Seat+"."+verb, in) if err != nil { @@ -38,55 +39,48 @@ func (k liveKeeper) verb(verb string, in, out any) error { return nil } -// Get subscribes to the chunks of a call of its own before it asks snapshot-get, so none is missed, then receives them in -// order, acknowledging each (§681, A-139). +// Get asks snapshot-get with a call of its own and receives its chunks in order through mesh-sdk's one receive of a chunked +// seat answer (process.Conn.CallChunks: subscribed before it asks, each chunk acknowledged; §681, A-139, design 47 G A-291). func (k liveKeeper) Get(ctx context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error) { var bits [16]byte if _, err := rand.Read(bits[:]); err != nil { return meshbroker.SnapshotGetAnswer{}, err } - send := buscalls.SeatChunks{Seat: meshbroker.Seat, Verb: meshbroker.VerbSnapshotGet, Call: hex.EncodeToString(bits[:])} - chunks := make(chan *nats.Msg, 4) - sub, err := k.nc.ChanSubscribe(send.Chunks(), chunks) - if err != nil { - return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the chunks of snapshot-get cannot be heard: %w", err) - } - defer func() { _ = sub.Unsubscribe() }() + call := hex.EncodeToString(bits[:]) var ans meshbroker.SnapshotGetAnswer - if err := k.verb(meshbroker.VerbSnapshotGet, meshbroker.SnapshotGetInput{Call: send.Call, Request: request}, &ans); err != nil { - return ans, err - } - if ans.Error != "" { - return ans, nil - } - if err := fits(ans); err != nil { - return ans, err - } - if ans.Size > 0 && ans.Chunks == 0 { - return ans, fmt.Errorf("the keeper answered the request %s's %d bytes with no chunk: this account is granted no chunk subjects of snapshot-get", request, ans.Size) - } - next := func() (string, []byte, error) { - bound := time.NewTimer(buscalls.SeatAckWithin) - defer bound.Stop() - select { - case m := <-chunks: - return m.Subject, m.Data, nil - case <-bound.C: - return "", nil, buscalls.ErrSeatChunkMissed - case <-ctx.Done(): - return "", nil, ctx.Err() + answered, receiving := false, false + accept := func(raw json.RawMessage) error { + answered = true + if err := json.Unmarshal(raw, &ans); err != nil { + return fmt.Errorf("%s's %s answered what does not read: %w", meshbroker.Seat, meshbroker.VerbSnapshotGet, err) } + if ans.Error != "" { + return errAnswered + } + if err := fits(ans); err != nil { + return err + } + if ans.Size > 0 && ans.Chunks == 0 { + return fmt.Errorf("the keeper answered the request %s's %d bytes with no chunk: this account is granted no chunk subjects of snapshot-get", request, ans.Size) + } + receiving = true + return nil } - ack := func(subject string) error { - _, err := k.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin) - return err - } - if _, err := send.Receive(ans.Chunks, next, ack, w); err != nil { + _, err := k.chunks(ctx, "seat:"+meshbroker.Seat+"."+meshbroker.VerbSnapshotGet, call, meshbroker.SnapshotGetInput{Call: call, Request: request}, accept, w) + switch { + case errors.Is(err, errAnswered), err == nil: + return ans, nil + case !answered: + return ans, fmt.Errorf("%s's %s: %w", meshbroker.Seat, meshbroker.VerbSnapshotGet, err) + case receiving: return ans, fmt.Errorf("the snapshot of the request %s: %w", request, err) } - return ans, nil + return ans, err } +// errAnswered ends a snapshot-get whose answer names an error before any chunk is taken: the answer says it. +var errAnswered = errors.New("snapshot-get answered an error") + func (k liveKeeper) Restore(_ context.Context, request string, verify bool, why string) (meshbroker.SnapshotRestoreAnswer, error) { var ans meshbroker.SnapshotRestoreAnswer err := k.verb(meshbroker.VerbSnapshotRestore, meshbroker.SnapshotRestoreInput{Request: request, Verify: verify, Why: why}, &ans) diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 7b08ab52a..a713dfdc1 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -167,7 +167,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro node: account.Node, points: dir, running: running}, Digests: digests{conn.State("digests")}, Dump: dump, - Keeper: liveKeeper{nc: nc, call: conn.Call, restored: restored}, + Keeper: liveKeeper{call: conn.Call, chunks: conn.CallChunks, restored: restored}, Bounds: stateBounds{conn.State("bounds")}, Settings: recovery.Settings{Cap: cap, KeepDays: keepDays, FillingMark: int(mark), DumpWait: dumpWait, RestoreWait: restoreWait}, Findings: writer,