From 99c36d8ddfac22903b2c793ccb5b56a51d015cd3 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:39:04 +0100 Subject: [PATCH] G sessions: the runner serves a bundle's put and get sessions and states put-sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A value over buscalls.ChunkBytes is put by mesh/objects.put-chunk (written into the module's store as each chunk is taken, sealed on the last, dropped at its bound or a disorder) and got by mesh/objects.get-chunk after a chunked answer, for a state's object and a sent object alike; the 8 MiB BundleObjectMost cap is gone with the stateobjects constant. Builds: design 47 G §518, §345, §457, §490, §620, §622, A-21, A-25 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/bus/bindings_test.go | 18 ++ node-tools/internal/bus/sent.go | 39 ++- node-tools/internal/bus/sessions.go | 76 +++++ node-tools/internal/launch/launch.go | 2 +- node-tools/internal/runtime/announce_test.go | 2 +- node-tools/internal/runtime/busstate.go | 72 ++++- node-tools/internal/runtime/ledgers_test.go | 2 +- node-tools/internal/runtime/runtime.go | 3 + node-tools/internal/runtime/sent_cap_test.go | 21 -- node-tools/internal/runtime/sessions.go | 301 +++++++++++++++++++ node-tools/internal/runtime/sessions_test.go | 195 ++++++++++++ 11 files changed, 681 insertions(+), 50 deletions(-) create mode 100644 node-tools/internal/bus/sessions.go delete mode 100644 node-tools/internal/runtime/sent_cap_test.go create mode 100644 node-tools/internal/runtime/sessions.go create mode 100644 node-tools/internal/runtime/sessions_test.go diff --git a/node-tools/internal/bus/bindings_test.go b/node-tools/internal/bus/bindings_test.go index 81c25a0..d6409a4 100644 --- a/node-tools/internal/bus/bindings_test.go +++ b/node-tools/internal/bus/bindings_test.go @@ -3,6 +3,7 @@ package bus_test import ( "context" "errors" + "io" "strings" "testing" @@ -70,6 +71,23 @@ func TestABundleReachesItsOwnBoundObjectsAndStreamsOnly(t *testing.T) { if err != nil || got == nil || string(got.Data) != "what was put" || got.Info.Name != "a" { t.Fatalf("read back %+v, %v", got, err) } + // Design 47 G §518: a value over one chunk is streamed into the store and read back out of it, its description kept. + big := strings.Repeat("0123456789", 120<<10) + info, err := conn.ObjectsPutStream(ctx, "m", buscalls.ObjectsNameRequest{Objects: "files", Name: "big"}, "streamed", strings.NewReader(big)) + if err != nil || info.Size != uint64(len(big)) || info.Description != "streamed" { + t.Fatalf("streamed put %+v, %v", info, err) + } + opened, rd, found, err := conn.ObjectsOpen(ctx, "m", buscalls.ObjectsNameRequest{Objects: "files", Name: "big"}) + if err != nil || !found || opened.Size != uint64(len(big)) { + t.Fatalf("opened %+v %v %v", opened, found, err) + } + if back, _ := io.ReadAll(rd); string(back) != big { + t.Fatalf("read back %d bytes, want %d", len(back), len(big)) + } + _ = rd.Close() + if err := conn.ObjectsDelete("m", buscalls.ObjectsNameRequest{Objects: "files", Name: "big"}); err != nil { + t.Fatal(err) + } if listed, err := conn.ObjectsList("m", buscalls.ObjectsRequest{Objects: "files"}); err != nil || len(listed.Objects) != 1 { t.Fatalf("listed %+v, %v", listed, err) } diff --git a/node-tools/internal/bus/sent.go b/node-tools/internal/bus/sent.go index 4eecf27..e8242fa 100644 --- a/node-tools/internal/bus/sent.go +++ b/node-tools/internal/bus/sent.go @@ -17,24 +17,11 @@ import ( // bindings alone. It reads as this connection's account, which the bus lets read a sender's sent store only where the // grants part granted it (a bundle is held to what its runtime channel carries by the runtime call, not here), from the manifests' callers and consumers; a refusal says the store. func (c *Conn) SentGet(r buscalls.ObjectsSentGetRequest) (*buscalls.ObjectData, error) { - ref := stateobjects.Ref{Store: r.Store, Object: r.Object, Size: r.Size, Digest: r.Digest, Form: stateobjects.FormVersion} - if _, ok := stateobjects.ReadRef(stateobjects.Wrap(ref)); !ok { - return nil, errors.New("a sent object is named by the reference's store, object, size and sha256, as the reply or event carried them") - } - if !strings.HasSuffix(r.Store, catalogue.BusNameSeparator+catalogue.SentObjectsLocal) { - return nil, fmt.Errorf("%s is no sent store (a module's is `%s%s`): only a sent object is read by its reference", - r.Store, catalogue.BusNameSeparator, catalogue.SentObjectsLocal) - } - js, err := c.JetStream() - if err != nil { - return nil, err - } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() - obs, err := js.ObjectStore(ctx, r.Store) + obs, err := c.sentStore(ctx, r) if err != nil { - return nil, fmt.Errorf("the sent store %s cannot be read as %s: it is not on the bus, or the bus grants this account no "+ - "read of it (the manifests' callers and consumers are granted): %w", r.Store, c.cred.Module, err) + return nil, err } info, err := obs.GetInfo(ctx, r.Object) if errors.Is(err, jetstream.ErrObjectNotFound) { @@ -49,3 +36,25 @@ func (c *Conn) SentGet(r buscalls.ObjectsSentGetRequest) (*buscalls.ObjectData, } return &buscalls.ObjectData{Info: objectInfoOf(info), Data: data}, nil } + +// sentStore checks the reference r and opens the sender's sent store it names. +func (c *Conn) sentStore(ctx context.Context, r buscalls.ObjectsSentGetRequest) (jetstream.ObjectStore, error) { + ref := stateobjects.Ref{Store: r.Store, Object: r.Object, Size: r.Size, Digest: r.Digest, Form: stateobjects.FormVersion} + if _, ok := stateobjects.ReadRef(stateobjects.Wrap(ref)); !ok { + return nil, errors.New("a sent object is named by the reference's store, object, size and sha256, as the reply or event carried them") + } + if !strings.HasSuffix(r.Store, catalogue.BusNameSeparator+catalogue.SentObjectsLocal) { + return nil, fmt.Errorf("%s is no sent store (a module's is `%s%s`): only a sent object is read by its reference", + r.Store, catalogue.BusNameSeparator, catalogue.SentObjectsLocal) + } + js, err := c.JetStream() + if err != nil { + return nil, err + } + obs, err := js.ObjectStore(ctx, r.Store) + if err != nil { + return nil, fmt.Errorf("the sent store %s cannot be read as %s: it is not on the bus, or the bus grants this account no "+ + "read of it (the manifests' callers and consumers are granted): %w", r.Store, c.cred.Module, err) + } + return obs, nil +} diff --git a/node-tools/internal/bus/sessions.go b/node-tools/internal/bus/sessions.go new file mode 100644 index 0000000..283b5c8 --- /dev/null +++ b/node-tools/internal/bus/sessions.go @@ -0,0 +1,76 @@ +package bus + +import ( + "context" + "errors" + "fmt" + "io" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "github.com/nats-io/nats.go/jetstream" +) + +// The bus side of a bundle's put and get sessions (design 47 G §518, A-21, A-25): a put written into the module's object +// store as its chunks arrive and a get read out of it as they are asked, through the store's own chunking, so no more than +// one chunk is held in memory. The session itself is the runtime's (internal/runtime/sessions.go). + +// ObjectsPutStream stores the object r.Name of the module's object store r.Objects from in, read to its end, with the +// put's description (A-25); it answers once the object is sealed. A read error of in drops the object: nothing is left. +func (c *Conn) ObjectsPutStream(ctx context.Context, module string, r buscalls.ObjectsNameRequest, description string, in io.Reader) (buscalls.ObjectInfo, error) { + if r.Name == "" { + return buscalls.ObjectInfo{}, errors.New("an object is put under a name") + } + obs, _, err := c.objectStore(ctx, module, r.Objects) + if err != nil { + return buscalls.ObjectInfo{}, err + } + info, err := obs.Put(ctx, jetstream.ObjectMeta{Name: r.Name, Description: description}, in) + if err != nil { + _ = obs.Delete(context.WithoutCancel(ctx), r.Name) + return buscalls.ObjectInfo{}, err + } + return objectInfoOf(info), nil +} + +// ObjectsOpen is one object of the module's object store as a reader of its bytes beside its info; found is false when there +// is none. The caller closes the reader. +func (c *Conn) ObjectsOpen(ctx context.Context, module string, r buscalls.ObjectsNameRequest) (info buscalls.ObjectInfo, rd io.ReadCloser, found bool, err error) { + obs, _, err := c.objectStore(ctx, module, r.Objects) + if err != nil { + return info, nil, false, err + } + res, err := obs.Get(ctx, r.Name) + if errors.Is(err, jetstream.ErrObjectNotFound) { + return info, nil, false, nil + } + if err != nil { + return info, nil, false, err + } + i, err := res.Info() + if err != nil { + res.Close() + return info, nil, false, err + } + return objectInfoOf(i), res, true, nil +} + +// SentOpen is a sent object, read by its reference as SentGet reads it, as a reader of its bytes beside its info. +func (c *Conn) SentOpen(ctx context.Context, r buscalls.ObjectsSentGetRequest) (buscalls.ObjectInfo, io.ReadCloser, error) { + obs, err := c.sentStore(ctx, r) + if err != nil { + return buscalls.ObjectInfo{}, nil, err + } + res, err := obs.Get(ctx, r.Object) + if errors.Is(err, jetstream.ErrObjectNotFound) { + return buscalls.ObjectInfo{}, nil, fmt.Errorf("the sent object %s is not in %s: it is kept 7 days, then expires", r.Object, r.Store) + } + if err != nil { + return buscalls.ObjectInfo{}, nil, fmt.Errorf("the sent object %s of %s: %w", r.Object, r.Store, err) + } + i, err := res.Info() + if err != nil { + res.Close() + return buscalls.ObjectInfo{}, nil, err + } + return objectInfoOf(i), res, nil +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index d293a19..633c848 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -123,7 +123,7 @@ var busStateCalls = map[string]bool{ buscalls.ObjectsInfo: true, buscalls.StreamRead: true, buscalls.StreamFill: true, buscalls.WorkQueueSize: true, buscalls.WorkQueueDead: true, buscalls.LedgerApply: true, buscalls.LedgerGet: true, buscalls.LedgerHistory: true, buscalls.LedgerQuery: true, buscalls.LedgerRebuild: true, buscalls.LedgerReindex: true, buscalls.LedgerSnapshot: true, - buscalls.LedgerLag: true, + buscalls.LedgerLag: true, buscalls.ObjectsSentGet: true, buscalls.ObjectsPutChunk: true, buscalls.ObjectsGetChunk: true, } // busStateWatches are the watches of mesh-sdk buscalls, each with the method its changes are handed to the bundle by. diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index 441c81e..cc67d5b 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -140,7 +140,7 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) { t.Fatalf("ping of the instance: %v", err) } var ping micro.Ping - if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,seat-settle-delay" { + if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,put-sessions,seat-settle-delay" { t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err) } } diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index 1c9dacc..5ac8bbd 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -1,13 +1,16 @@ package runtime import ( + "context" "encoding/json" "fmt" + "io" "sort" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscalls" - "git.novox.be/novox/mesh-sdk/go/stateobjects" + + "github.com/novox/mesh-tools/node-tools/internal/bus" ) // The capabilities the tool runner states of the calls it answers (novox/hq ADR 0595 §9, ADR 0597; design 25 § 12.4): each is @@ -17,6 +20,8 @@ import ( const ( CapBusStateCalls = "bus-state-calls" CapLedgerCalls = "ledger-calls" + // CapPutSessions is the put and get sessions' word (design 47 G §518, §622), mesh-sdk's. + CapPutSessions = stdio.CapPutSessions ) // callFamilies are the methods each capability stands for, the ones BusState and BusStateWatch answer. @@ -25,6 +30,7 @@ var callFamilies = map[string][]string{ buscalls.ObjectsWatch, buscalls.ObjectsSentGet, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead}, CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery, buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag}, + CapPutSessions: {buscalls.ObjectsPutChunk, buscalls.ObjectsGetChunk}, } // CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat @@ -54,11 +60,10 @@ func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMes case buscalls.ObjectsGet: var r buscalls.ObjectsNameRequest if err = decode(method, params, &r); err == nil { - var got *buscalls.ObjectData - if got, err = conn.ObjectsGet(b.module, r); err == nil && got == nil { + answer, err = b.all.objectsGet(b.module, r) + if err == nil && answer == nil { return json.RawMessage("null"), nil } - answer = got } case buscalls.ObjectsDelete: var r buscalls.ObjectsNameRequest @@ -69,14 +74,12 @@ func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMes case buscalls.ObjectsSentGet: var r buscalls.ObjectsSentGetRequest if err = decode(method, params, &r); err == nil { - var got *buscalls.ObjectData - if r.Size > stateobjects.BundleObjectMost { - // Only the bundle's channel is limited (design 47 "Any size"): the console reads a larger one whole. - err = fmt.Errorf("the sent object %s is %d bytes and a bundle's runtime channel carries at most %d: it is not read through a bundle", r.Object, r.Size, stateobjects.BundleObjectMost) - } else if got, err = conn.SentGet(r); err == nil { - answer = got - } + answer, err = b.all.sentGet(b.module, r) } + case buscalls.ObjectsPutChunk: + answer, err = b.all.putChunk(b.module, params) + case buscalls.ObjectsGetChunk: + answer, err = b.all.getChunk(b.module, params) case buscalls.ObjectsList: var r buscalls.ObjectsRequest if err = decode(method, params, &r); err == nil { @@ -147,3 +150,50 @@ func decode(method string, params json.RawMessage, into any) error { } return nil } + +// objectsGet answers a get: the object whole when it is at most one chunk, else the session that reads it chunk by chunk +// (design 47 G §490, §518); nil when there is none. +func (c *consumers) objectsGet(module string, r buscalls.ObjectsNameRequest) (any, error) { + st := c.objects() + if st == nil { + return nil, fmt.Errorf("the runner has no bus to read %s of %s from", r.Name, module) + } + ctx, cancel := context.WithTimeout(context.Background(), bus.StateTimeout) + defer cancel() + info, rd, found, err := st.ObjectsOpen(ctx, module, r) + if err != nil || !found { + return nil, err + } + if info.Size > buscalls.ChunkBytes { + return c.openGet(module, info, rd, sessionDigest(info.Digest), false) + } + defer rd.Close() + data, err := io.ReadAll(rd) + if err != nil { + return nil, err + } + return &buscalls.ObjectData{Info: info, Data: data}, nil +} + +// sentGet answers a get of a sent object by its reference: whole when the reference's size is at most one chunk, else a +// session (design 47 G §620). +func (c *consumers) sentGet(module string, r buscalls.ObjectsSentGetRequest) (any, error) { + if r.Size <= buscalls.ChunkBytes { + got, err := c.conn.SentGet(r) + if err != nil { + return nil, err + } + return got, nil + } + st := c.objects() + if st == nil { + return nil, fmt.Errorf("the runner has no bus to read %s from", r.Object) + } + ctx, cancel := context.WithTimeout(context.Background(), bus.StateTimeout) + defer cancel() + info, rd, err := st.SentOpen(ctx, r) + if err != nil { + return nil, err + } + return c.openGet(module, info, rd, sessionDigest(r.Digest), true) +} diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 805984a..96bc96e 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/node-tools/internal/runtime/ledgers_test.go @@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) { // no other name; none answers "the runtime answers no ...". func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) { m, _, _ := aLedger(t) - if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,seat-settle-delay" { + if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,put-sessions,seat-settle-delay" { t.Fatalf("the runner states %s", got) } for capability, methods := range callFamilies { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 84580ef..f3bc178 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -770,6 +770,8 @@ type consumers struct { ledgers ledgerSource // calls are the tool calls in flight, by the id each bundle was handed (ledgers.go, novox/hq ADR 0597 §7). calls callsInFlight + // sessions are the bundles' put and get sessions (sessions.go). + sessions sessionTable } type moduleEvents struct { @@ -845,6 +847,7 @@ func (c *consumers) dropModule(module string) { } } c.provisioned.stopModule(module) + c.dropModuleSessions(module) } // forModule is the bus one launched bundle of a module reaches the mesh through. diff --git a/node-tools/internal/runtime/sent_cap_test.go b/node-tools/internal/runtime/sent_cap_test.go deleted file mode 100644 index 1498a39..0000000 --- a/node-tools/internal/runtime/sent_cap_test.go +++ /dev/null @@ -1,21 +0,0 @@ -package runtime - -import ( - "encoding/json" - "strings" - "testing" - - "git.novox.be/novox/mesh-sdk/go/buscalls" - "git.novox.be/novox/mesh-sdk/go/stateobjects" -) - -// Design 47 G §69, §62, §407: only a bundle's runtime channel holds the cap, so the runtime call refuses a sent object over -// it before the bus is asked, where the console (which reads through the bus) reads it whole. -func TestARuntimeCallRefusesASentObjectOverTheBundleChannel(t *testing.T) { - params, _ := json.Marshal(buscalls.ObjectsSentGetRequest{Store: "m_sent-objects", Object: "o", - Size: int64(stateobjects.BundleObjectMost) + 1, Digest: strings.Repeat("a", 64)}) - _, err := (&moduleBus{all: &consumers{}}).BusState(buscalls.ObjectsSentGet, params) - if err == nil || !strings.Contains(err.Error(), "not read through a bundle") { - t.Fatalf("a sent object over the channel was answered %v", err) - } -} diff --git a/node-tools/internal/runtime/sessions.go b/node-tools/internal/runtime/sessions.go new file mode 100644 index 0000000..7227f64 --- /dev/null +++ b/node-tools/internal/runtime/sessions.go @@ -0,0 +1,301 @@ +package runtime + +import ( + "bufio" + "context" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "github.com/novox/mesh-tools/node-tools/internal/launch" +) + +// A bundle's put and get sessions (design 47 G §518, A-21, A-25, §620): a value over buscalls.ChunkBytes crosses the +// bundle's channel as calls the runner holds the session of, nothing of it on the bus. A put writes each chunk into the +// module's object store as it arrives (through a pipe into the store's own chunking, so no more than one chunk is held) and +// seals the object on the last chunk; a get reads the object out as the bundle asks its chunks, in order. + +// SessionBound is how long a session has, counted once for the whole session from its first chunk (§518 "The bound"): the +// call's bound, launch.CallTimeout. +var SessionBound = launch.CallTimeout + +// objectSessions is what a session needs of the bus: *bus.Conn. +type objectSessions interface { + ObjectsPutStream(ctx context.Context, module string, r buscalls.ObjectsNameRequest, description string, in io.Reader) (buscalls.ObjectInfo, error) + ObjectsOpen(ctx context.Context, module string, r buscalls.ObjectsNameRequest) (buscalls.ObjectInfo, io.ReadCloser, bool, error) + SentOpen(ctx context.Context, r buscalls.ObjectsSentGetRequest) (buscalls.ObjectInfo, io.ReadCloser, error) +} + +// sessionTable holds the sessions of every module, each under its module's name and the bundle's call id, so a bundle +// reaches its own sessions alone. +type sessionTable struct { + mu sync.Mutex + puts map[string]*putSession + gets map[string]*getSession + // store replaces the bus where a test has none. + store objectSessions +} + +type putSession struct { + mu sync.Mutex + next uint32 + pw *io.PipeWriter + done chan struct{} // closed once res is set + res putResult + cancel context.CancelFunc + timer *time.Timer + store string +} + +type putResult struct { + info buscalls.ObjectInfo + err error +} + +type getSession struct { + mu sync.Mutex + next uint32 + rd *bufio.Reader + closer io.Closer + timer *time.Timer +} + +var errSessionBound = errors.New("the session was not completed within the call's bound") + +func sessionKey(module, call string) string { return module + " " + call } + +// putChunk takes one chunk of a put session: it answers {seq}, or {error}, and the put's own answer for the last chunk. +func (c *consumers) putChunk(module string, params json.RawMessage) (any, error) { + var r buscalls.ObjectsPutChunkRequest + if err := decode(buscalls.ObjectsPutChunk, params, &r); err != nil { + return nil, err + } + if r.Call == "" { + return nil, errors.New("a chunk names the put's call") + } + t := &c.sessions + key := sessionKey(module, r.Call) + t.mu.Lock() + s := t.puts[key] + if r.Seq == 1 && s == nil { + var err error + if s, err = c.beginPut(module, key, r); err != nil { + t.mu.Unlock() + return buscalls.ObjectsChunkTaken{Error: err.Error()}, nil + } + } + t.mu.Unlock() + if s == nil { + return buscalls.ObjectsChunkTaken{Error: fmt.Sprintf("no put %s of %s is open at chunk %d: it was dropped at its bound, or never begun", r.Call, r.Name, r.Seq)}, nil + } + s.mu.Lock() + defer s.mu.Unlock() + if r.Seq != s.next { + c.dropPut(key, s, errors.New("a chunk out of order")) + return buscalls.ObjectsChunkTaken{Error: fmt.Sprintf("putting %s: chunk %d came where chunk %d was next", r.Name, r.Seq, s.next)}, nil + } + if len(r.Data) > buscalls.ChunkBytes { + c.dropPut(key, s, errors.New("a chunk over the limit")) + return buscalls.ObjectsChunkTaken{Error: fmt.Sprintf("putting %s: chunk %d is %d bytes, over the %d a chunk carries", r.Name, r.Seq, len(r.Data), buscalls.ChunkBytes)}, nil + } + if _, err := s.pw.Write(r.Data); err != nil { + // The store refused the chunk (its cap) or the bound passed: its own reason is the answer. + <-s.done + res := s.res + c.forgetPut(key, s) + reason := res.err + if reason == nil || errors.Is(reason, io.ErrClosedPipe) { + reason = err + } + return buscalls.ObjectsChunkTaken{Error: fmt.Sprintf("putting %s into %s, at chunk %d: %v", r.Name, s.store, r.Seq, reason)}, nil + } + if !r.Last { + s.next++ + return buscalls.ObjectsChunkTaken{Seq: r.Seq}, nil + } + _ = s.pw.Close() + <-s.done + res := s.res + c.forgetPut(key, s) + if res.err != nil { + return buscalls.ObjectsChunkTaken{Error: fmt.Sprintf("putting %s into %s: %v", r.Name, s.store, res.err)}, nil + } + return res.info, nil +} + +// beginPut opens the put of the first chunk, with t.mu held. +func (c *consumers) beginPut(module, key string, r buscalls.ObjectsPutChunkRequest) (*putSession, error) { + if r.Name == "" { + return nil, errors.New("an object is put under a name") + } + st := c.objects() + if st == nil { + return nil, errors.New("the runner has no bus to put on") + } + pr, pw := io.Pipe() + ctx, cancel := context.WithCancel(context.Background()) + s := &putSession{next: 1, pw: pw, done: make(chan struct{}), cancel: cancel, store: r.Objects} + s.timer = time.AfterFunc(SessionBound, func() { + c.sessions.mu.Lock() + current := c.sessions.puts[key] == s + c.sessions.mu.Unlock() + if current { + _ = pw.CloseWithError(errSessionBound) + cancel() + } + }) + if c.sessions.puts == nil { + c.sessions.puts = map[string]*putSession{} + } + c.sessions.puts[key] = s + go func() { + info, err := st.ObjectsPutStream(ctx, module, buscalls.ObjectsNameRequest{Objects: r.Objects, Name: r.Name}, r.Description, pr) + _ = pr.CloseWithError(err) + s.res = putResult{info, err} + close(s.done) + }() + return s, nil +} + +// dropPut ends a put that will not be sealed: the store purges what it wrote. +func (c *consumers) dropPut(key string, s *putSession, why error) { + _ = s.pw.CloseWithError(why) + <-s.done + c.forgetPut(key, s) +} + +func (c *consumers) forgetPut(key string, s *putSession) { + s.timer.Stop() + s.cancel() + c.sessions.mu.Lock() + delete(c.sessions.puts, key) + c.sessions.mu.Unlock() +} + +// objects is the bus a session reads and writes objects through. +func (c *consumers) objects() objectSessions { + if c.sessions.store != nil { + return c.sessions.store + } + if c.conn == nil { + return nil + } + return c.conn +} + +// openGet registers the get session of an opened object and answers the chunked answer that begins it. +func (c *consumers) openGet(module string, info buscalls.ObjectInfo, rd io.ReadCloser, digest string, sent bool) (any, error) { + call := newCall() + g := &getSession{next: 1, rd: bufio.NewReaderSize(rd, buscalls.ChunkBytes), closer: rd} + key := sessionKey(module, call) + g.timer = time.AfterFunc(SessionBound, func() { c.dropGet(key, g) }) + c.sessions.mu.Lock() + if c.sessions.gets == nil { + c.sessions.gets = map[string]*getSession{} + } + c.sessions.gets[key] = g + c.sessions.mu.Unlock() + var chunked json.RawMessage = json.RawMessage("true") + if !sent { + chunked = json.RawMessage(fmt.Sprint((info.Size + buscalls.ChunkBytes - 1) / buscalls.ChunkBytes)) + } + return buscalls.ObjectsChunked{Chunked: chunked, Size: info.Size, Digest: digest, Call: call}, nil +} + +func (c *consumers) dropGet(key string, g *getSession) { + g.timer.Stop() + c.sessions.mu.Lock() + delete(c.sessions.gets, key) + c.sessions.mu.Unlock() + _ = g.closer.Close() +} + +// getChunk answers the next chunk of a get session: {seq, last, data}, or {error}. +func (c *consumers) getChunk(module string, params json.RawMessage) (any, error) { + var r buscalls.ObjectsGetChunkRequest + if err := decode(buscalls.ObjectsGetChunk, params, &r); err != nil { + return nil, err + } + key := sessionKey(module, r.Call) + c.sessions.mu.Lock() + g := c.sessions.gets[key] + c.sessions.mu.Unlock() + if g == nil { + return buscalls.ObjectsChunk{Error: fmt.Sprintf("no get %s is open at chunk %d: it was dropped at its bound, or never begun", r.Call, r.Seq)}, nil + } + g.mu.Lock() + defer g.mu.Unlock() + if r.Seq != g.next { + c.dropGet(key, g) + return buscalls.ObjectsChunk{Error: fmt.Sprintf("chunk %d was asked where chunk %d is next", r.Seq, g.next)}, nil + } + buf := make([]byte, buscalls.ChunkBytes) + n, err := io.ReadFull(g.rd, buf) + last := errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) + if err != nil && !last { + c.dropGet(key, g) + return buscalls.ObjectsChunk{Error: fmt.Sprintf("reading chunk %d: %v", r.Seq, err)}, nil + } + if !last { + if _, perr := g.rd.Peek(1); errors.Is(perr, io.EOF) { + last = true + } + } + g.next++ + if last { + c.dropGet(key, g) + } + return buscalls.ObjectsChunk{Seq: r.Seq, Last: last, Data: buf[:n]}, nil +} + +// dropModuleSessions ends every session of a module whose bundle is gone. +func (c *consumers) dropModuleSessions(module string) { + c.sessions.mu.Lock() + var puts []*putSession + var putKeys []string + var gets []*getSession + var getKeys []string + for k, s := range c.sessions.puts { + if strings.HasPrefix(k, module+" ") { + puts, putKeys = append(puts, s), append(putKeys, k) + } + } + for k, g := range c.sessions.gets { + if strings.HasPrefix(k, module+" ") { + gets, getKeys = append(gets, g), append(getKeys, k) + } + } + c.sessions.mu.Unlock() + for i, s := range puts { + c.dropPut(putKeys[i], s, errors.New("the bundle's channel closed")) + } + for i, g := range gets { + c.dropGet(getKeys[i], g) + } +} + +// sessionDigest is the `sha256:` form of an object's digest as the bus store keeps it (`SHA-256=` and base64), or as a +// reference carries it (hex). +func sessionDigest(d string) string { + if raw, ok := strings.CutPrefix(d, "SHA-256="); ok { + if b, err := base64.URLEncoding.DecodeString(raw); err == nil { + return "sha256:" + hex.EncodeToString(b) + } + } + return "sha256:" + strings.TrimPrefix(d, "sha256:") +} + +func newCall() string { + var b [12]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} diff --git a/node-tools/internal/runtime/sessions_test.go b/node-tools/internal/runtime/sessions_test.go new file mode 100644 index 0000000..5924c97 --- /dev/null +++ b/node-tools/internal/runtime/sessions_test.go @@ -0,0 +1,195 @@ +package runtime + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// fakeObjects is the bus of a session: what was put whole, what a read gives, and what the store deleted. +type fakeObjects struct { + put map[string][]byte + description string + deleted []string + read []byte + ctxEnded chan struct{} +} + +func (f *fakeObjects) ObjectsPutStream(ctx context.Context, _ string, r buscalls.ObjectsNameRequest, description string, in io.Reader) (buscalls.ObjectInfo, error) { + data, err := io.ReadAll(in) + if err != nil { + f.deleted = append(f.deleted, r.Name) + if f.ctxEnded != nil { + <-ctx.Done() + close(f.ctxEnded) + } + return buscalls.ObjectInfo{}, err + } + f.put[r.Name], f.description = data, description + return buscalls.ObjectInfo{Name: r.Name, Size: uint64(len(data)), Description: description}, nil +} + +func (f *fakeObjects) ObjectsOpen(context.Context, string, buscalls.ObjectsNameRequest) (buscalls.ObjectInfo, io.ReadCloser, bool, error) { + return buscalls.ObjectInfo{Name: "o", Size: uint64(len(f.read))}, io.NopCloser(bytes.NewReader(f.read)), true, nil +} + +func (f *fakeObjects) SentOpen(context.Context, buscalls.ObjectsSentGetRequest) (buscalls.ObjectInfo, io.ReadCloser, error) { + return buscalls.ObjectInfo{Size: uint64(len(f.read))}, io.NopCloser(bytes.NewReader(f.read)), nil +} + +func aSession(t *testing.T, f *fakeObjects) *moduleBus { + t.Helper() + c := &consumers{} + c.sessions.store = f + return &moduleBus{all: c, module: "m"} +} + +func asked(t *testing.T, b *moduleBus, method string, req any, into any) { + t.Helper() + params, _ := json.Marshal(req) + raw, err := b.BusState(method, params) + if err != nil { + t.Fatalf("%s: %v", method, err) + } + if err := json.Unmarshal(raw, into); err != nil { + t.Fatal(err) + } +} + +// Design 47 G §518, A-21, A-25: a put session writes each chunk as it is taken, carries the description of its first chunk, +// and answers the put's own info for the last; a chunk out of order fails the put naming both sequences and leaves none. +func TestAPutSessionSealsOnTheLastChunkAndDropsOnADisorder(t *testing.T) { + f := &fakeObjects{put: map[string][]byte{}} + b := aSession(t, f) + whole := bytes.Repeat([]byte("x"), 2*buscalls.ChunkBytes+5) + var taken buscalls.ObjectsChunkTaken + for seq, at := uint32(1), 0; ; seq++ { + end := min(at+buscalls.ChunkBytes, len(whole)) + req := buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "a", Call: "c1", Seq: seq, Last: end == len(whole), Data: whole[at:end]} + if seq == 1 { + req.Description = "what it is" + } + if req.Last { + var info buscalls.ObjectInfo + asked(t, b, buscalls.ObjectsPutChunk, req, &info) + if info.Size != uint64(len(whole)) || info.Description != "what it is" { + t.Fatalf("the put answered %+v", info) + } + break + } + asked(t, b, buscalls.ObjectsPutChunk, req, &taken) + if taken.Seq != seq || taken.Error != "" { + t.Fatalf("chunk %d was answered %+v", seq, taken) + } + at = end + } + if !bytes.Equal(f.put["a"], whole) { + t.Fatalf("the store holds %d bytes, want %d", len(f.put["a"]), len(whole)) + } + + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 1, Data: []byte("one")}, &taken) + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 3, Data: []byte("three")}, &taken) + if !strings.Contains(taken.Error, "chunk 3") || !strings.Contains(taken.Error, "chunk 2") { + t.Fatalf("a disorder was answered %+v", taken) + } + if _, ok := f.put["b"]; ok || len(f.deleted) != 1 || f.deleted[0] != "b" { + t.Fatalf("a dropped put left %v, deleted %v", f.put, f.deleted) + } + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 2, Data: []byte("two")}, &taken) + if taken.Error == "" { + t.Fatal("a chunk of a dropped put was taken") + } +} + +// §518 "The bound": a put not sealed within the bound is dropped, its object deleted. +func TestAPutNotSealedWithinTheBoundIsDropped(t *testing.T) { + was := SessionBound + SessionBound = time.Millisecond + t.Cleanup(func() { SessionBound = was }) + f := &fakeObjects{put: map[string][]byte{}, ctxEnded: make(chan struct{})} + b := aSession(t, f) + var taken buscalls.ObjectsChunkTaken + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "a", Call: "c", Seq: 1, Data: []byte("x")}, &taken) + <-f.ctxEnded + if len(f.deleted) != 1 { + t.Fatalf("the unsealed object was left: %v", f.deleted) + } +} + +// §490, §518, §620: a value over one chunk is got as a session (the chunked answer, then each chunk by its call and sequence, +// the whole checked against the digest); a sent object over one chunk the same, with `chunked` true. +func TestAGetOverOneChunkRunsAsASession(t *testing.T) { + whole := bytes.Repeat([]byte("y"), 2*buscalls.ChunkBytes) + f := &fakeObjects{read: whole} + b := aSession(t, f) + sum := sha256.Sum256(whole) + want := "sha256:" + hex.EncodeToString(sum[:]) + for name, run := range map[string]func() buscalls.ObjectsChunked{ + "state": func() buscalls.ObjectsChunked { + var c buscalls.ObjectsChunked + asked(t, b, buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: "files", Name: "o"}, &c) + return c + }, + "sent": func() buscalls.ObjectsChunked { + var c buscalls.ObjectsChunked + asked(t, b, buscalls.ObjectsSentGet, buscalls.ObjectsSentGetRequest{Store: "m_sent-objects", Object: "o", Size: int64(len(whole)), Digest: hex.EncodeToString(sum[:])}, &c) + if string(c.Chunked) != "true" { + t.Errorf("a sent object's chunked is %s", c.Chunked) + } + return c + }, + } { + t.Run(name, func(t *testing.T) { + c := run() + if c.Call == "" || c.Size != uint64(len(whole)) { + t.Fatalf("the get answered %+v", c) + } + if name == "state" { + if string(c.Chunked) != "2" { + t.Errorf("chunked is %s, want 2", c.Chunked) + } + } + var got []byte + for seq := uint32(1); ; seq++ { + var ch buscalls.ObjectsChunk + asked(t, b, buscalls.ObjectsGetChunk, buscalls.ObjectsGetChunkRequest{Call: c.Call, Seq: seq}, &ch) + if ch.Error != "" || ch.Seq != seq { + t.Fatalf("chunk %d: %+v", seq, ch) + } + got = append(got, ch.Data...) + if ch.Last { + if seq != 2 { + t.Fatalf("an object of two chunks ended at chunk %d", seq) + } + break + } + } + if !bytes.Equal(got, whole) || name == "state" && c.Digest == "" { + t.Fatalf("read %d bytes, digest %s", len(got), c.Digest) + } + if name == "sent" && c.Digest != want { + t.Errorf("digest %s, want %s", c.Digest, want) + } + var again buscalls.ObjectsChunk + asked(t, b, buscalls.ObjectsGetChunk, buscalls.ObjectsGetChunkRequest{Call: c.Call, Seq: 3}, &again) + if again.Error == "" { + t.Error("a finished session answered another chunk") + } + }) + } + // A value of one chunk or less is the value itself, as before. + f.read = []byte("small") + var small buscalls.ObjectData + asked(t, b, buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: "files", Name: "o"}, &small) + if string(small.Data) != "small" { + t.Fatalf("a small value was answered %+v", small) + } +}