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 69e3c1d4..77aa3eee 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -25,6 +25,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -693,6 +694,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hub.Handle(meshwants.SubjectWaitEnded, func(m jetstream.Msg) { heardWaitEnded(ctx, ix, m, logger.Printf) }); err != nil { return err } + if err := hub.Handle(nodebuildgo.SubjectStepEndedAll, func(m jetstream.Msg) { heardStepEnded(ctx, ix, m, logger.Printf) }); err != nil { + return err + } if err := hub.Handle(decidedSubject(cred.Module), func(m jetstream.Msg) { heardDecided(ctx, registry, m, logger.Printf) }); err != nil { return err } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go index ce127695..1f0cdcd5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go @@ -15,6 +15,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-indexer/internal/index" @@ -112,6 +113,22 @@ func heardWaitEnded(ctx context.Context, ix *index.Indexer, m jetstream.Msg, log _ = m.Ack() } +// heardStepEnded deletes the input kept for the run ask a build seat's step-ended names (design 47 G §354, A-78 d). +func heardStepEnded(ctx context.Context, ix *index.Indexer, m jetstream.Msg, logf func(string, ...any)) { + var e nodebuildgo.StepEnded + if err := json.Unmarshal(m.Data(), &e); err != nil { + logf("%s holds what is no step-ended, left: %v", m.Subject(), err) + _ = m.Ack() + return + } + if err := ix.HearInputEnded(ctx, e); err != nil { + logf("inputs: the input of the ask %s could not be deleted, so the bus gives it again: %v", e.Ask, err) + _ = m.Nak() + return + } + _ = m.Ack() +} + // sayVersionsFilled says versions-filled (design 47 G §372) once, on the seat's stream that keeps it with no age limit. func sayVersionsFilled(conn *nats.Conn, cred process.Account) func(context.Context, meshcatalog.VersionsFilledEvent, string) error { return func(_ context.Context, e meshcatalog.VersionsFilledEvent, id string) error { diff --git a/modules/mesh-delivery-indexer/internal/index/run_inputs.go b/modules/mesh-delivery-indexer/internal/index/run_inputs.go index 647737ae..ccb5070b 100644 --- a/modules/mesh-delivery-indexer/internal/index/run_inputs.go +++ b/modules/mesh-delivery-indexer/internal/index/run_inputs.go @@ -9,10 +9,13 @@ import ( "errors" "fmt" "slices" + "strings" "time" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" ) // The indexer's inputs (design 47 G §354): the input of the ask `run` the indexer's validation puts, kept in the old state @@ -28,6 +31,9 @@ const ( // InputsSwept is how many of the oldest keys each new ask judges. InputsSwept = 20 inputsTime = "20060102T150405Z" + // InputsAskPrefix opens the keys that name an input by the id of the run ask it was kept for (A-78 d): ask., the id escaped as a + // name (§73), holding the input's own key, so the build seat's step-ended of that ask finds it and deletes it. + InputsAskPrefix = "ask." ) // InputDigest is the sha256 of an input's exact bytes, as "sha256:" and its hex digits. @@ -36,9 +42,13 @@ func InputDigest(body []byte) string { return "sha256:" + hex.EncodeToString(sum[:]) } -// KeepInput keeps the exact bytes of a run ask's input under their write time and sha256, after sweeping the keys more than a -// day old among the 20 oldest; it answers the digest. The same bytes written again in the same second are one key. -func (ix *Indexer) KeepInput(ctx context.Context, body []byte) (string, error) { +// 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. +func (ix *Indexer) KeepInput(ctx context.Context, ask string, body []byte) (string, error) { + if ask == "" { + return "", errors.New("name the run ask the input is kept for") + } if ix.Inputs == nil { return "", errors.New("the state inputs is not kept here") } @@ -54,21 +64,53 @@ func (ix *Indexer) KeepInput(ctx context.Context, body []byte) (string, error) { if err := ix.Inputs.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) { return "", err } + if err := ix.Inputs.Create(ctx, InputsAskPrefix+keys.EscapeName(ask), []byte(key)); err != nil && !errors.Is(err, ErrConflict) { + return "", err + } return digest, nil } +// HearInputEnded deletes the input kept for a run ask when the build seat's step-ended of that ask is heard (§354, A-78 d); an ask +// that kept none, or whose input a sweep deleted, changes nothing. +func (ix *Indexer) HearInputEnded(ctx context.Context, e nodebuildgo.StepEnded) error { + if ix.Inputs == nil || e.Ask == "" { + return nil + } + askKey := InputsAskPrefix + keys.EscapeName(e.Ask) + raw, rev, found, err := ix.Inputs.Get(ctx, askKey) + if err != nil || !found { + return err + } + if _, irev, ok, err := ix.Inputs.Get(ctx, string(raw)); err != nil { + return err + } else if ok { + if err := ix.Inputs.Delete(ctx, string(raw), irev); err != nil { + return err + } + } + return ix.Inputs.Delete(ctx, askKey, rev) +} + // sweepInputs deletes, among the 20 oldest keys, those written more than a day ago. func (ix *Indexer) sweepInputs(ctx context.Context) error { all, err := ix.Inputs.Keys(ctx) if err != nil { return err } - slices.Sort(all) // the write time leads the key, so the oldest come first - if len(all) > InputsSwept { - all = all[:InputsSwept] - } var errs []error + var inputs, asks []string for _, k := range all { + if strings.HasPrefix(k, InputsAskPrefix) { + asks = append(asks, k) + } else { + inputs = append(inputs, k) + } + } + slices.Sort(inputs) // the write time leads the key, so the oldest come first + if len(inputs) > InputsSwept { + inputs = inputs[:InputsSwept] + } + for _, k := range inputs { t, _, ok := meshcatalog.ParseInputsByTimeDigestKey(k) at, perr := time.Parse(inputsTime, t) if !ok || perr != nil || ix.now().Sub(at) < InputsKeptFor { @@ -80,6 +122,16 @@ func (ix *Indexer) sweepInputs(ctx context.Context) error { } } } + // a name by ask whose input a sweep or its step-ended deleted goes with it + for _, k := range asks { + if raw, rev, found, err := ix.Inputs.Get(ctx, k); err == nil && found { + if _, _, ok, err := ix.Inputs.Get(ctx, string(raw)); err == nil && !ok { + if err := ix.Inputs.Delete(ctx, k, rev); err != nil { + errs = append(errs, err) + } + } + } + } return errors.Join(errs...) } @@ -90,7 +142,7 @@ func InputKey(ctx context.Context, st Store, digest string) (string, error) { return "", err } for _, k := range all { - if _, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k); ok && d == digest { + if _, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k); ok && d == digest && !strings.HasPrefix(k, InputsAskPrefix) { return k, nil } } @@ -162,7 +214,7 @@ func InputsReadTool(st Store) stdio.Tool { out := InputsAnswer{Inputs: []InputKept{}} for _, k := range all { _, d, ok := meshcatalog.ParseInputsByTimeDigestKey(k) - if !ok || (want != "" && d != want) { + if !ok || strings.HasPrefix(k, InputsAskPrefix) || (want != "" && d != want) { continue } raw, _, found, err := st.Get(ctx, k) diff --git a/modules/mesh-delivery-indexer/internal/index/versions_test.go b/modules/mesh-delivery-indexer/internal/index/versions_test.go index 224982dc..8920edf1 100644 --- a/modules/mesh-delivery-indexer/internal/index/versions_test.go +++ b/modules/mesh-delivery-indexer/internal/index/versions_test.go @@ -15,6 +15,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" ) // Design 47 G §349: a version is released once every artifact the module's entry names is published at it, said once as @@ -255,7 +256,7 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) { ctx := context.Background() now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) ix := &Indexer{Inputs: newMem(), Now: func() time.Time { return now }} - d1, err := ix.KeepInput(ctx, []byte("first")) + d1, err := ix.KeepInput(ctx, "ask-1", []byte("first")) if err != nil { t.Fatal(err) } @@ -267,18 +268,30 @@ func TestAnInputIsKeptByItsDigestAndSweptAfterADay(t *testing.T) { t.Error("an input never kept is answered") } now = now.Add(23 * time.Hour) - if _, err := ix.KeepInput(ctx, []byte("second")); err != nil { + if _, err := ix.KeepInput(ctx, "ask-2", []byte("second")); err != nil { t.Fatal(err) } - if keys, _ := ix.Inputs.Keys(ctx); len(keys) != 2 { + inputsOnly := func() []string { + var out []string + all, _ := ix.Inputs.Keys(ctx) + for _, k := range all { + if !strings.HasPrefix(k, InputsAskPrefix) { + out = append(out, k) + } + } + return out + } + if keys := inputsOnly(); len(keys) != 2 { t.Fatalf("inside a day: %v", keys) } now = now.Add(2 * time.Hour) - d3, _ := ix.KeepInput(ctx, []byte("third")) - keys, _ := ix.Inputs.Keys(ctx) - if len(keys) != 2 { + d3, _ := ix.KeepInput(ctx, "ask-3", []byte("third")) + if keys := inputsOnly(); len(keys) != 2 { t.Fatalf("after a day the first stays: %v", keys) } + 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 { t.Error("the input a day old is still answered") } @@ -483,3 +496,37 @@ func TestTheFirstFillGivesTheNewestVersionsTheirFilledEdges(t *testing.T) { t.Errorf("outdated after the first fill: %+v, %v", a.Behind, err) } } + +// Design 47 G §354, A-78 d: an input is also keyed by its run ask's id, so the build seat's step-ended of that ask deletes it, and +// the key by ask no other read counts as an input. +func TestAnInputIsDeletedAtTheStepEndedOfItsAsk(t *testing.T) { + ctx := context.Background() + now := time.Date(2026, 11, 1, 3, 0, 0, 0, time.UTC) + ix := &Indexer{Inputs: newMem(), Now: func() time.Time { return now }} + d1, err := ix.KeepInput(ctx, "a.1", []byte("first")) + if err != nil { + t.Fatal(err) + } + d2, _ := ix.KeepInput(ctx, "a.2", []byte("second")) + if _, err := ix.KeepInput(ctx, "", []byte("x")); err == nil { + t.Error("an input kept for no ask") + } + if err := ix.HearInputEnded(ctx, nodebuildgo.StepEnded{Ask: "a.1"}); err != nil { + t.Fatal(err) + } + 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 { + t.Error("the input of an ended ask is still answered") + } + if _, err := Input(ctx, ix.Inputs, meshcatalog.InputInput{Digest: 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 { + 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 c7c45db8..37cd2b28 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -161,7 +161,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // The registry seats' published fill each (§413: no consume of version.numbered) package's series (DLV-I5, design 51 *The versions*). if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", "mesh-delivery-checker.preview.written", "mesh-delivery-checker.try.ended", "mesh-delivery-checker.try.started", "mesh-delivery-checker.verdict", "mesh-delivery-forge-bridge.forge.**", - "mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + "mesh-delivery.source-moved", "mesh-wants.version-delivered", "mesh-wants.wait-ended", "node-build-go.step-ended", "node-engine.running", "npm-package-registry.published", "operator-channel.decided"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index dde0c85e..2058d2cc 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -86,6 +86,7 @@ "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-wants.version-delivered", + "node-build-go.step-ended", "node-engine.running", "mesh-wants.wait-ended", "mesh-delivery-checker.verdict",