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

indexer: an input is also keyed by its run ask's id and deleted at the build seat's step-ended of that ask

Browse Source 
KeepInput takes the ask's id. It has no caller yet: the indexer publishes no run ask for its validation in this tree.

Builds: design 47 G §354, A-78 d
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 23:19:29 +01:00
parent bdc9a61524
commit 64def921ea
6 changed files with 137 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+4
View File
@@ -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
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/versions.go
+17
View File
@@ -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 {
modules/mesh-delivery-indexer/internal/index/run_inputs.go
+61 -9
View File
@@ -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.<id>, 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)
modules/mesh-delivery-indexer/internal/index/versions_test.go
+53 -6
View File
@@ -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)
}
}
modules/mesh-delivery-indexer/manifest_test.go
+1 -1
View File
@@ -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" {
modules/mesh-delivery-indexer/module.json
+1
View File
@@ -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",