diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index 1509c9c2..c767b78b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -10,6 +10,7 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/stateobjects" "os" "strings" "sync" @@ -291,7 +292,7 @@ func main() { // the parts whose entry declares it (novox/hq ADR 0663). tools = append(tools, index.RevisionTool(revisions, index.Module), index.EntryTool(revisions, keptRead{stdio.State(index.EntryIndexState)}), index.EntriesTool(revisions, keptRead{stdio.State(index.EntryIndexState)}, keptRead{stdio.State(index.NamedState)}), index.VersionsTool(keptRead{stdio.State(index.VersionsState)}), index.OutdatedTool(&index.Indexer{Versions: keptRead{stdio.State(index.VersionsState)}, Heard: keptRead{stdio.State(index.HeardState)}, Revisions: revisions}), - index.InputTool(keptRead{stdio.State(index.InputsState)}), index.InputsReadTool(keptRead{stdio.State(index.InputsState)})) + index.InputsReadTool(keptRead{stdio.State(index.InputsState)}, stateobjects.ForState(index.InputsState).Objects)) tools = append(tools, index.CatalogRevisionTools(revisions, asks, time.Now, func(what, why string) error { _, err := acts.Record(meshhandactsclient.Request{What: what, Why: why, Cause: index.DropCandidateGraphTool, Verb: "drop-candidate-graph"}) return err diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go new file mode 100644 index 00000000..91048da9 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/input_integration_test.go @@ -0,0 +1,95 @@ +//go:build integration + +package main + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "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/catalogue" + "git.novox.be/novox/mesh-sdk/go/stateobjects" + "mesh-delivery-indexer/internal/index" +) + +// TestIndexerAnswersARunsInputOverOneMiBInChunks (design 47 G §712, A-237, §681): an input of 3 MiB, kept in the state inputs with +// its object beside it, is answered by the seat mesh-catalog's verb input as a count of chunks and then sent on the call's chunk +// subjects, each acknowledged by the caller; the chunks carry the exact bytes. +func TestIndexerAnswersARunsInputOverOneMiBInChunks(t *testing.T) { + body := make([]byte, 3<<20) + if _, err := rand.Read(body); err != nil { + t.Fatal(err) + } + sum := sha256.Sum256(body) + digest := hex.EncodeToString(sum[:]) + js, _, _, _ := startIndexer(t, func(context.Context, buildLog) {}, func(ctx context.Context, js jetstream.JetStream) { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.InputsBucket, History: 1}) + if err != nil { + t.Fatal(err) + } + objs, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: catalogue.BusName(index.Module, catalogue.StateObjectsName(index.InputsState))}) + if err != nil { + t.Fatal(err) + } + key := meshcatalog.InputsByTimeDigestKey(time.Now().UTC().Format("20060102T150405Z"), digest) + if err := stateobjects.OverBus(kv, objs, catalogue.StateObjectsName(index.InputsState)).Put(key, base64.StdEncoding.EncodeToString(body)); err != nil { + t.Fatal(err) + } + }) + nc := js.Conn() + call := "0123456789abcdef0123" + send := buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: call} + chunks := make(chan *nats.Msg, 16) + sub, err := nc.ChanSubscribe(send.Chunks(), chunks) + if err != nil { + t.Fatal(err) + } + defer sub.Unsubscribe() + req, _ := json.Marshal(meshcatalog.InputInput{Digest: digest, Call: call}) + var reply struct { + Result meshcatalog.InputAnswer `json:"result"` + Error string `json:"error"` + } + for deadline := time.Now().Add(30 * time.Second); ; time.Sleep(100 * time.Millisecond) { + m, err := nc.Request("mesh.seat.mesh-catalog.tool.input", req, 5*time.Second) + if err == nil { + if err := json.Unmarshal(m.Data, &reply); err != nil { + t.Fatal(err) + } + break + } + if time.Now().After(deadline) { + t.Fatalf("the verb input is not answered within 30 s: %v", err) + } + } + if reply.Error != "" || reply.Result.Digest != digest || reply.Result.Chunks != 6 { + t.Fatalf("the answer %+v (%q), want %s in 6 chunks", reply.Result, reply.Error, digest) + } + var got bytes.Buffer + next := func() (string, []byte, error) { + select { + case m := <-chunks: + return m.Subject, m.Data, nil + case <-time.After(20 * time.Second): + return "", nil, buscalls.ErrSeatChunkMissed + } + } + ack := func(subject string) error { _, err := nc.Request(subject, []byte("{}"), 5*time.Second); return err } + if _, err := send.Receive(reply.Result.Chunks, next, ack, &got); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got.Bytes(), body) { + t.Errorf("the chunks carry %d bytes that are not the input's %d", got.Len(), len(body)) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go new file mode 100644 index 00000000..91ff9258 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/inputs.go @@ -0,0 +1,98 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/stateobjects" + "mesh-delivery-indexer/internal/index" +) + +// inputObjects is the object store beside the state inputs (`inputs-objects`, a state's own, design 47 G §712, A-237), opened when +// first used: until the grants part has made it every call is process.ErrNotYet. +type inputObjects struct { + js jetstream.JetStream + module string + mu sync.Mutex + bus stateobjects.Objects +} + +func (o *inputObjects) open() (stateobjects.Objects, error) { + o.mu.Lock() + defer o.mu.Unlock() + if o.bus != nil { + return o.bus, nil + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + store, err := o.js.ObjectStore(ctx, catalogue.BusName(o.module, catalogue.StateObjectsName(index.InputsState))) + if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) { + return nil, fmt.Errorf("the object store of the state inputs is not on the bus yet: %w", process.ErrNotYet) + } + if err != nil { + return nil, fmt.Errorf("the object store of the state inputs is not open: %w", err) + } + o.bus = stateobjects.BusObjects{Store: store} + return o.bus, nil +} + +func (o *inputObjects) MostBytes() int { return 0 } + +func (o *inputObjects) Put(name string, data []byte) error { + b, err := o.open() + if err != nil { + return err + } + return b.Put(name, data) +} + +func (o *inputObjects) Get(name string) ([]byte, error) { + b, err := o.open() + if err != nil { + return nil, err + } + return b.Get(name) +} + +func (o *inputObjects) Delete(name string) error { + b, err := o.open() + if err != nil { + return err + } + return b.Delete(name) +} + +func (o *inputObjects) List() ([]stateobjects.Object, error) { + b, err := o.open() + if err != nil { + return nil, err + } + 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 a9fca35f..cd82847e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -23,9 +23,11 @@ import ( "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/buscalls" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -443,6 +445,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro pinsBucket := process.AwaitBucket(ctx, js, conn, index.PinsBucket, "no record's pin on a catalog graph revision is kept, so no revision is removed", logger.Printf) pinsStore := awaitedStore{b: pinsBucket} consumers := process.AwaitBucket(ctx, js, conn, index.Module+"_"+process.ConsumersState, "indexer_consumer reads no consumer", logger.Printf) + // The state inputs and the object store beside it; the seat mesh-catalog's verb input is answered here in chunks of the core + // connection, of any size (design 47 G §712, A-237). + 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)}) + if err != nil { + return fmt.Errorf("answering the verb input: %w", err) + } + defer stopInput() ix := &index.Indexer{NamedSets: named, Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, @@ -458,7 +470,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Versions: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.VersionsBucket, "no released version is kept, so no package-published makes a version", logger.Printf)}, BuildEdges: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.BuildEdgesBucket, "no build of another watched branch keeps its edges", logger.Printf)}, Pins: pinsStore, PinsPending: pendingOf(js, cred), - Inputs: 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)}, + Inputs: inputsStore, + InputObjects: inObjects, SayVersionReleased: sayVersionReleased(conn, cred), SayVersionsFilled: sayVersionsFilled(conn, cred), // The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear; // until then every merge is taken in at once, as before. diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 2c2898fa..d55ddff0 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -6,6 +6,7 @@ import ( "context" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/stateobjects" "reflect" "slices" "strings" @@ -120,6 +121,9 @@ type Indexer struct { SayVersionsFilled func(ctx context.Context, e meshcatalog.VersionsFilledEvent, id string) error // Inputs is the state inputs (run_inputs.go, design 47 G §354): the inputs of the indexer's run asks; nil keeps none. Inputs Store + // InputObjects is the object store beside the state inputs (`inputs-objects`): an input too large for a key is kept there + // (mesh-sdk stateobjects, design 47 G §712, A-237); nil keeps a large input nowhere and refuses it. + InputObjects stateobjects.Objects // Registry is the registrations and their trusted changes (registrations.go, design 47 G §2 to §4, §667); SeedModules reads // what the first registrations are composed from (§8). Nil: no registration act is answered. Registry *Registry diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 22a18610..ae1aee31 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -910,7 +910,7 @@ func (ix *Indexer) dry() *dryRun { Requirements: nil, Findings: nil, SeatDefinitions: nil, ReadFile: ix.ReadFile, SaySeatDefinition: nil, OwnBuildState: nil, Heard: nil, Placement: nil, RunningWhole: nil, underSeats: true, // A dry run keeps no catalog graph revision (catalog_revisions.go): its revision is the take-in's to keep. - Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, NamedSets: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil, Pins: nil, PinsPending: nil, + Revisions: nil, SayCatalogRevision: nil, SayFill: nil, EntryIndex: nil, NamedSets: nil, Versions: nil, SayVersionReleased: nil, SayVersionsFilled: nil, Inputs: nil, InputObjects: nil, Pins: nil, PinsPending: nil, // A dry run writes no registration and asks no one. Registry: nil, SeedModules: nil} x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index ba497782..a3e072a5 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -1131,7 +1131,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "Unjudged", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", - "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "BuildEdges", "Pins", "PinsPending"} { + "Requirements", "Findings", "SeatDefinitions", "ReadFile", "SaySeatDefinition", "Runs", "OwnBuildState", "Heard", "Placement", "RunningWhole", "Revisions", "SayCatalogRevision", "SayFill", "EntryIndex", "NamedSets", "Registry", "SeedModules", "Versions", "SayVersionReleased", "SayVersionsFilled", "Inputs", "InputObjects", "BuildEdges", "Pins", "PinsPending"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) 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 new file mode 100644 index 00000000..bee9315a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/run_input_chunks_test.go @@ -0,0 +1,132 @@ +package index + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "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" +) + +type objectsMem struct { + mu sync.Mutex + objs map[string][]byte +} + +func (*objectsMem) MostBytes() int { return 0 } +func (o *objectsMem) Put(name string, data []byte) error { + o.mu.Lock() + defer o.mu.Unlock() + o.objs[name] = data + return nil +} +func (o *objectsMem) Get(name string) ([]byte, error) { + o.mu.Lock() + defer o.mu.Unlock() + d, ok := o.objs[name] + if !ok { + return nil, errors.New("no such object") + } + return d, nil +} +func (o *objectsMem) Delete(name string) error { return nil } +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) + chunks map[string][]byte +} + +func (l *lineMem) Publish(subject string, data []byte) error { + l.mu.Lock() + l.chunks[subject] = data + h := l.hear + l.mu.Unlock() + var c buscalls.SeatChunk + 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)) + return nil +} + +func (l *lineMem) Hear(_ string, fn func(string)) (func(), error) { + l.mu.Lock() + l.hear = fn + l.mu.Unlock() + return func() {}, 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 +// input answers it in chunks of at most 512 KiB, each acknowledged before the next. +func TestAnInputOverOneMiBIsKeptAsAnObjectAndAnsweredInChunks(t *testing.T) { + ctx := context.Background() + objs := &objectsMem{objs: map[string][]byte{}} + ix := &Indexer{Inputs: newMem(), InputObjects: objs, Now: time.Now} + body := make([]byte, 3<<20+17) + if _, err := rand.Read(body); err != nil { + t.Fatal(err) + } + digest, err := ix.KeepInput(ctx, "ask-big", body) + if err != nil { + t.Fatal(err) + } + if len(objs.objs) != 1 { + t.Fatalf("a large input is one object beside the state, got %d", len(objs.objs)) + } + if got, err := InputBody(ctx, ix.Inputs, objs, digest); err != nil || !bytes.Equal(got, body) { + t.Fatalf("the input read back: %d bytes, %v", len(got), err) + } + line := &lineMem{chunks: map[string][]byte{}} + verb := InputVerb(ix.Inputs, objs, line, time.Second, t.Logf) + raw, _ := json.Marshal(meshcatalog.InputInput{Digest: digest, Call: "0123456789abcdef"}) + out, err := verb("", raw) + if err != nil { + t.Fatal(err) + } + ans := out.(meshcatalog.InputAnswer) + want := int64((len(body) + buscalls.SeatChunkBytes - 1) / buscalls.SeatChunkBytes) + if ans.Digest != digest || ans.Chunks != want || want < 6 { + t.Fatalf("the answer %+v, want %d chunks", ans, want) + } + send := buscalls.SeatChunks{Seat: meshcatalog.Seat, Verb: meshcatalog.VerbInput, Call: "0123456789abcdef"} + var got []byte + deadline := time.Now().Add(5 * time.Second) + for n := int64(1); n <= want; n++ { + for { + line.mu.Lock() + data, ok := line.chunks[send.Chunk(n)] + line.mu.Unlock() + if ok { + var c buscalls.SeatChunk + if err := json.Unmarshal(data, &c); err != nil || c.N != n || c.Of != want { + t.Fatalf("chunk %d: %+v, %v", n, c, err) + } + got = append(got, c.Data...) + break + } + if time.Now().After(deadline) { + t.Fatalf("chunk %d of %d never sent", n, want) + } + time.Sleep(time.Millisecond) + } + } + if !bytes.Equal(got, body) { + t.Error("the chunks do not carry the input's bytes") + } + if _, err := verb("", json.RawMessage(`{"digest":"sha256:none","call":"0123456789abcdef"}`)); err == nil { + t.Error("an input never kept is answered") + } + if _, err := verb("", json.RawMessage(`{"digest":"`+digest+`","call":"x"}`)); err == nil { + t.Error("a call of no token is answered") + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/run_inputs.go b/modules/mesh-delivery-indexer/internal/index/run_inputs.go index 236fa226..5ecfd042 100644 --- a/modules/mesh-delivery-indexer/internal/index/run_inputs.go +++ b/modules/mesh-delivery-indexer/internal/index/run_inputs.go @@ -8,6 +8,7 @@ import ( "encoding/json" "errors" "fmt" + "regexp" "slices" "strings" "time" @@ -15,7 +16,10 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/stateobjects" ) // The indexer's inputs (design 47 G §354): the input of the ask `run` the indexer's validation puts, kept in the old state @@ -42,6 +46,13 @@ func InputDigest(body []byte) string { return "sha256:" + hex.EncodeToString(sum[:]) } +// runDigest is the digest a run ask names its input by (buildstep.RunInput.Digest): the 64 hex digits of its sha256, which are also +// the digest of the key's form, a name a state's key may hold (a ':' may not be). +func runDigest(body []byte) string { + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) +} + // KeepInput keeps the exact bytes of the input of the run ask ask under their write time and sha256, after sweeping the keys more // than a day old among the 20 oldest, and names it by the ask's id (A-78 d); it answers the digest. The same bytes written again in // the same second are one key. @@ -55,13 +66,9 @@ func (ix *Indexer) KeepInput(ctx context.Context, ask string, body []byte) (stri if err := ix.sweepInputs(ctx); err != nil { ix.logf("inputs: the keys older than a day are not all deleted: %v", err) } - digest := InputDigest(body) + digest := runDigest(body) key := meshcatalog.InputsByTimeDigestKey(ix.now().UTC().Format(inputsTime), digest) - data, err := json.Marshal(base64.StdEncoding.EncodeToString(body)) - if err != nil { - return "", err - } - if err := ix.Inputs.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) { + if err := inputValues(ix.Inputs, ix.InputObjects).Put(key, base64.StdEncoding.EncodeToString(body)); err != nil && !errors.Is(err, stateobjects.ErrChanged) { return "", err } if err := ix.Inputs.Create(ctx, InputsAskPrefix+keys.EscapeName(ask), []byte(key)); err != nil && !errors.Is(err, ErrConflict) { @@ -149,40 +156,132 @@ func InputKey(ctx context.Context, st Store, digest string) (string, error) { return "", nil } -// Input answers the verb input: the exact bytes (base64) of the input of the digest, its digest checked. -func Input(ctx context.Context, st Store, in meshcatalog.InputInput) (meshcatalog.InputAnswer, error) { - var out meshcatalog.InputAnswer - if in.Digest == "" { - return out, errors.New("name the input's sha256 (digest)") - } - key, err := InputKey(ctx, st, in.Digest) - if err != nil { - return out, fmt.Errorf("the state %s cannot be read: %w", InputsState, err) - } - raw, _, found, err := st.Get(ctx, key) - if key == "" || err != nil || !found { - return out, fmt.Errorf("no input of digest %s is kept", in.Digest) - } - text, err := meshcatalog.DecodeInputsByTimeDigestValue(raw) - if err != nil { - return out, fmt.Errorf("the input %s cannot be read: %w", key, err) - } - body, err := base64.StdEncoding.DecodeString(text) - if err != nil || InputDigest(body) != in.Digest { - return out, fmt.Errorf("the input kept under %s does not match its digest", key) - } - return meshcatalog.InputAnswer{Digest: in.Digest, Input: text}, nil +// inputValues is the state inputs with its object store beside it (mesh-sdk stateobjects): a value that fits a key is in it, a +// larger one is an object with its reference in the key (ADR 0666 §16). +func inputValues(st Store, objs stateobjects.Objects) stateobjects.Values { + return stateobjects.Values{KV: storeKV{st}, Objects: objs, Store: catalogue.StateObjectsName(InputsState)} } -// InputTool is the verb input as the tool mesh-catalog.input of the indexer's own runtime. -func InputTool(st Store) stdio.Tool { - return stdio.Tool{Name: meshcatalog.Seat + "." + meshcatalog.VerbInput, Run: func(args map[string]any) (any, error) { +// storeKV is a Store as the state's key-value side. +type storeKV struct{ s Store } + +func (k storeKV) Get(key string) (*stateobjects.Entry, error) { + v, rev, found, err := k.s.Get(context.Background(), key) + if err != nil || !found { + return nil, err + } + return &stateobjects.Entry{Value: v, Revision: rev}, nil +} + +func changed(err error) error { + if errors.Is(err, ErrConflict) { + return stateobjects.ErrChanged + } + return err +} + +func (k storeKV) Create(key string, v []byte) error { + return changed(k.s.Create(context.Background(), key, v)) +} + +func (k storeKV) Update(key string, v []byte, rev uint64) error { + return changed(k.s.Update(context.Background(), key, v, rev)) +} + +func (k storeKV) Delete(key string) error { + _, rev, found, err := k.s.Get(context.Background(), key) + if err != nil || !found { + return err + } + return k.s.Delete(context.Background(), key, rev) +} + +func (k storeKV) Keys() ([]string, error) { return k.s.Keys(context.Background()) } + +// InputBody is the exact bytes of the input of the digest, checked against it, whatever its size. +func InputBody(ctx context.Context, st Store, objs stateobjects.Objects, digest string) ([]byte, error) { + if digest == "" { + return nil, errors.New("name the input's sha256 (digest)") + } + key, err := InputKey(ctx, st, digest) + if err != nil { + return nil, fmt.Errorf("the state %s cannot be read: %w", InputsState, err) + } + if key == "" { + return nil, fmt.Errorf("no input of digest %s is kept", digest) + } + var text string + found, err := inputValues(st, objs).Get(key, &text) + if err != nil { + return nil, fmt.Errorf("the input %s cannot be read: %w", key, err) + } + if !found { + return nil, fmt.Errorf("no input of digest %s is kept", digest) + } + body, err := base64.StdEncoding.DecodeString(text) + if err != nil || runDigest(body) != digest { + return nil, fmt.Errorf("the input kept under %s does not match its digest", key) + } + 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) { + return func(_ string, raw json.RawMessage) (any, error) { var in meshcatalog.InputInput - if raw, err := json.Marshal(args); err == nil { - _ = json.Unmarshal(raw, &in) + if err := json.Unmarshal(raw, &in); err != nil { + return nil, fmt.Errorf("the arguments are not JSON: %w", err) } - return Input(context.Background(), st, in) - }} + if !callToken.MatchString(in.Call) { + return nil, errors.New("call is one token of 16 to 64 letters, digits, - or _: 128 random bits the caller chooses") + } + body, err := InputBody(context.Background(), st, objs, in.Digest) + if err != nil { + return nil, err + } + 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) + } + 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 + } + } + }() + return meshcatalog.InputAnswer{Digest: in.Digest, Chunks: total}, nil + } } // InputsAnswer is what indexer_inputs answers: each key kept with its time, digest and size, or one input's bytes. @@ -200,7 +299,7 @@ type InputKept struct { // InputsReadTool is indexer_inputs (class read): every input the indexer keeps, with its key, digest and size; with a digest, // that input whole (base64). -func InputsReadTool(st Store) stdio.Tool { +func InputsReadTool(st Store, objs stateobjects.Objects) stdio.Tool { return stdio.Tool{Name: InputsTool, Description: "Read the indexer's inputs of the ask run (design 47 G section 354): each key (write time and sha256) with its size; with a digest, that input's exact bytes, base64.", Input: map[string]any{"type": "object", "properties": map[string]any{"digest": map[string]any{"type": "string", "description": "one input's sha256; none lists every key"}}}, Run: func(args map[string]any) (any, error) { @@ -221,9 +320,18 @@ func InputsReadTool(st Store) stdio.Tool { if err != nil || !found { continue } - text, _ := meshcatalog.DecodeInputsByTimeDigestValue(raw) - kept := InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(len(text))} + var kept InputKept + if ref, isRef := stateobjects.ReadRef(raw); isRef { + kept = InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(int(ref.Size) - 2)} // the quotes of the JSON string + } else { + text, _ := meshcatalog.DecodeInputsByTimeDigestValue(raw) + kept = InputKept{Key: k, Digest: d, Bytes: base64.StdEncoding.DecodedLen(len(text))} + } if want != "" { + var text string + if ok, err := inputValues(st, objs).Get(k, &text); err != nil || !ok { + continue + } kept.Input = text } out.Inputs = append(out.Inputs, kept) @@ -231,3 +339,19 @@ func InputsReadTool(st Store) 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/internal/index/versions_test.go b/modules/mesh-delivery-indexer/internal/index/versions_test.go index f81918b5..61598e8d 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_test.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_test.go @@ -260,11 +260,10 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) { if err != nil { t.Fatal(err) } - a, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}) - if err != nil || a.Digest != d1 || a.Input != "Zmlyc3Q=" { - t.Fatalf("the input read back: %+v, %v", a, err) + if a, err := InputBody(ctx, ix.Inputs, nil, d1); err != nil || string(a) != "first" { + t.Fatalf("the input read back: %q, %v", a, err) } - if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: "sha256:none"}); err == nil { + if _, err := InputBody(ctx, ix.Inputs, nil, "sha256:none"); err == nil { t.Error("an input never kept is answered") } now = now.Add(23 * time.Hour) @@ -292,10 +291,10 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) { if all, _ := ix.Inputs.Keys(ctx); slices.Contains(all, InputsAskPrefix+"ask-1") { t.Errorf("the name by ask of the swept input stays: %v", all) } - if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}); err == nil { + if _, err := InputBody(ctx, ix.Inputs, nil, d1); err == nil { t.Error("the input a day old is still answered") } - if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d3}); err != nil { + if _, err := InputBody(ctx, ix.Inputs, nil, d3); err != nil { t.Error(err) } } @@ -517,16 +516,16 @@ func TestAnInputIsDeletedAtTheStepEndedOfItsAsk(t *testing.T) { if err := ix.HearInputEnded(ctx, nodebuildgo.StepEnded{Ask: "a.1"}); err != nil { t.Errorf("an end heard again: %v", err) } - if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d1}); err == nil { + if _, err := InputBody(ctx, ix.Inputs, nil, d1); err == nil { t.Error("the input of an ended ask is still answered") } - if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: d2}); err != nil { + if _, err := InputBody(ctx, ix.Inputs, nil, d2); err != nil { t.Errorf("the input of another ask: %v", err) } if got, _ := ix.Inputs.Keys(ctx); len(got) != 2 { t.Errorf("keys after one end: %v", got) } - if a, err := InputsReadTool(ix.Inputs).Run(map[string]any{}); err != nil || len(a.(InputsAnswer).Inputs) != 1 { + if a, err := InputsReadTool(ix.Inputs, nil).Run(map[string]any{}); err != nil || len(a.(InputsAnswer).Inputs) != 1 { t.Errorf("indexer_inputs counts a key by ask: %+v, %v", a, err) } } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 2eba8cea..d7da242e 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -382,7 +382,9 @@ 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. - if f == "Subscribe(" && strings.HasSuffix(path, "seatlist.go") { + // 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")) { continue } if strings.Contains(string(src), f) {