From 87edff67eeb50665567c4a02bb680778d48739c7 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:27:01 +0100 Subject: [PATCH] B5g5-1: the indexer sends the input's chunks with buscalls.SeatChunks.Send, so its Subscribe exception goes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The acknowledgement subscription is the SDK's; the indexer's inputs.go holds none, and the test that forbids a subscription in the program no longer excepts it. Builds: design 47 G §712, §681, A-237, A-139 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-indexer/inputs.go | 20 ------- .../cmd/mesh-delivery-indexer/main.go | 2 +- .../internal/index/run_input_chunks_test.go | 10 ++-- .../internal/index/run_inputs.go | 53 ++----------------- .../mesh-delivery-indexer/manifest_test.go | 4 +- 5 files changed, 12 insertions(+), 77 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go index 91ff9258..3f872366 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go @@ -7,7 +7,6 @@ import ( "sync" "time" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -77,22 +76,3 @@ func (o *inputObjects) List() ([]stateobjects.Object, error) { } return b.List() } - -// coreChunks is the core connection as a send of chunks reaches it (index.ChunkLine): a chunk is published on no stream, an -// acknowledgement is a request the holder answers. -type coreChunks struct{ nc *nats.Conn } - -func (c coreChunks) Publish(subject string, data []byte) error { return c.nc.Publish(subject, data) } - -func (c coreChunks) Hear(subject string, fn func(string)) (func(), error) { - sub, err := c.nc.Subscribe(subject, func(m *nats.Msg) { - fn(m.Subject) - if m.Reply != "" { - _ = m.Respond([]byte("{}")) - } - }) - if err != nil { - return nil, err - } - return func() { _ = sub.Unsubscribe() }, nil -} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index cd82847e..8b0e1394 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -450,7 +450,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro inputsStore := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.InputsBucket, "no input of a run ask is kept, so the verb input answers none", logger.Printf)} inObjects := &inputObjects{js: js, module: cred.Module} stopInput, err := calls.AnswerSeat(meshcatalog.Seat, map[string]process.SeatHandler{ - meshcatalog.VerbInput: index.InputVerb(inputsStore, inObjects, coreChunks{conn}, buscalls.SeatAckWithin, logger.Printf)}) + meshcatalog.VerbInput: index.InputVerb(inputsStore, inObjects, conn, buscalls.SeatAckWithin, logger.Printf)}) if err != nil { return fmt.Errorf("answering the verb input: %w", err) } diff --git a/modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go b/modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go index bee9315a..120f2ff8 100644 --- a/modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go +++ b/modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go @@ -10,6 +10,8 @@ import ( "testing" "time" + "github.com/nats-io/nats.go" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/stateobjects" @@ -42,7 +44,7 @@ func (o *objectsMem) List() ([]stateobjects.Object, error) { return nil, nil } // lineMem is a core bus in memory: a chunk published is kept, and its acknowledgement heard at once. type lineMem struct { mu sync.Mutex - hear func(string) + hear nats.MsgHandler chunks map[string][]byte } @@ -55,15 +57,15 @@ func (l *lineMem) Publish(subject string, data []byte) error { if err := json.Unmarshal(data, &c); err != nil { return err } - go h(buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: "0123456789abcdef"}.Ack(c.N)) + go h(&nats.Msg{Subject: buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: "0123456789abcdef"}.Ack(c.N)}) return nil } -func (l *lineMem) Hear(_ string, fn func(string)) (func(), error) { +func (l *lineMem) Subscribe(_ string, fn nats.MsgHandler) (*nats.Subscription, error) { l.mu.Lock() l.hear = fn l.mu.Unlock() - return func() {}, nil + return nil, nil } // Design 47 G §712, A-237: an input is of any size: over a key's limit it is kept as an object beside the state, and the verb diff --git a/modules/mesh-delivery-indexer/internal/index/run_inputs.go b/modules/mesh-delivery-indexer/internal/index/run_inputs.go index 5ecfd042..8a93c64d 100644 --- a/modules/mesh-delivery-indexer/internal/index/run_inputs.go +++ b/modules/mesh-delivery-indexer/internal/index/run_inputs.go @@ -225,20 +225,12 @@ func InputBody(ctx context.Context, st Store, objs stateobjects.Objects, digest return body, nil } -// ChunkLine is the core bus as a send of chunks reaches it: a chunk is published on its subject, kept on no stream, and each is -// acknowledged by the caller as a request on its ack subject, which the holder answers (design 47 G §681). -type ChunkLine interface { - Publish(subject string, data []byte) error - // Hear calls fn with the subject of each request made on subject (a pattern) and answers it, until stop is called. - Hear(subject string, fn func(subject string)) (stop func(), err error) -} - var callToken = regexp.MustCompile(`^[A-Za-z0-9_-]{16,64}$`) // InputVerb answers the verb input (marked chunks): the digest and how many chunks carry the bytes, then sends them in chunks of at // most buscalls.SeatChunkBytes on the call's chunk subjects, each acknowledged before the next (A-237, A-139). The input is of any // size. A caller that does not acknowledge within buscalls.SeatAckWithin ends the send, said in logf. -func InputVerb(st Store, objs stateobjects.Objects, line ChunkLine, within time.Duration, logf func(string, ...any)) func(caller string, body json.RawMessage) (any, error) { +func InputVerb(st Store, objs stateobjects.Objects, line buscalls.ChunkLine, within time.Duration, logf func(string, ...any)) func(caller string, body json.RawMessage) (any, error) { return func(_ string, raw json.RawMessage) (any, error) { var in meshcatalog.InputInput if err := json.Unmarshal(raw, &in); err != nil { @@ -253,31 +245,10 @@ func InputVerb(st Store, objs stateobjects.Objects, line ChunkLine, within time. } send := buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: in.Call} total := int64((len(body) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes) - acks := make(chan string, 1) - // heard before the answer goes, since the caller acknowledges only after it - stop, err := line.Hear(buscalls.SeatAcks(meshcatalog.Seat, meshcatalog.VerbInput), func(subject string) { - select { - case acks <- subject: - default: - } - }) - if err != nil { - return nil, fmt.Errorf("the acknowledgements cannot be heard, so no chunk is sent: %w", err) - } + // the send subscribes to the acknowledgements itself and ends when it has sent every chunk or one is not acknowledged go func() { - defer stop() - for n := int64(1); n <= total; n++ { - from := (n - 1) * buscalls.SeatChunkBytes - to := min(int64(len(body)), from+buscalls.SeatChunkBytes) - data, _ := json.Marshal(buscalls.SeatChunk{N: n, Of: total, Data: body[from:to]}) - if err := line.Publish(send.Chunk(n), data); err != nil { - logf("the send %s of the input %s ends at chunk %d of %d: %v", in.Call, in.Digest, n, total, err) - return - } - if !acknowledged(acks, send.Ack(n), within) { - logf("the send %s of the input %s ends at chunk %d of %d: no acknowledgement within %s", in.Call, in.Digest, n, total, within) - return - } + if err := send.Send(line, body, within, logf); err != nil { + logf("the send %s of the input %s: %v", in.Call, in.Digest, err) } }() return meshcatalog.InputAnswer{Digest: in.Digest, Chunks: total}, nil @@ -339,19 +310,3 @@ func InputsReadTool(st Store, objs stateobjects.Objects) stdio.Tool { return out, nil }} } - -// acknowledged waits within for the acknowledgement on subject, skipping another call's. -func acknowledged(acks <-chan string, subject string, within time.Duration) bool { - ctx, cancel := context.WithTimeout(context.Background(), within) - defer cancel() - for { - select { - case got := <-acks: - if got == subject { - return true - } - case <-ctx.Done(): - return false - } - } -} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index d7da242e..2eba8cea 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -382,9 +382,7 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1238/AC9 (ADR for _, f := range forbidden { // The seat list reader's Bus interface names a method Subscribe; the indexer's own bus hands it its own events // directly and subscribes to nothing on the bus. - // The acknowledgements of the chunks of the verb input are heard on the core connection while a send runs (design 47 G - // §712, A-237, §681): cmd/mesh-delivery-indexer/inputs.go. - if f == "Subscribe(" && (strings.HasSuffix(path, "seatlist.go") || strings.HasSuffix(path, "mesh-delivery-indexer/inputs.go")) { + if f == "Subscribe(" && strings.HasSuffix(path, "seatlist.go") { continue } if strings.Contains(string(src), f) {