novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

B5g5-1: the indexer sends the input's chunks with buscalls.SeatChunks.Send, so its Subscribe exception goes

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-12 09:27:01 +01:00
parent 2aed3151af
commit 87edff67ee
5 changed files with 12 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go
-20
View File
@@ -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
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+1 -1
View File
@@ -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)
}
modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go
+6 -4
View File
@@ -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
modules/mesh-delivery-indexer/internal/index/run_inputs.go
+4 -49
View File
@@ -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
}
}
}
modules/mesh-delivery-indexer/manifest_test.go
+1 -3
View File
@@ -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) {