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

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)

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,