From 409c41d4aeef215da3929f08c8a882238fadd2c2 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 00:30:52 +0000 Subject: [PATCH] Stamp the modules with the gate's source revision, read from source.main (issue 1344) --- .../cmd/mesh-delivery-indexer/main.go | 26 +++++++++++++++--- .../cmd/mesh-delivery-indexer/main_test.go | 21 +++++++++++++++ .../cmd/mesh-delivery-indexer/source.go | 27 +++++++++++++++++++ modules/mesh-delivery-indexer/go.mod | 2 +- modules/mesh-delivery-indexer/go.sum | 2 ++ .../internal/index/candidates.go | 4 +++ .../internal/index/modules.go | 6 +++-- 7 files changed, 82 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go 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 5465157..a64e005 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -255,9 +255,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - // The gate's source bucket (mesh-delivery-gate_source) is read from release 3d on; it does not exist before, and the - // indexer does not wait for it. - known := index.NewBuilds() each := func(e jetstream.KeyValueEntry) { if opOf(e) { @@ -274,6 +271,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { errs <- follow(ctx, w, each) }() ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Builds: known, Logf: logger.Printf} + // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key + // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does + // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. + sourceErrs := make(chan error, 1) + switch source, err := js.KeyValue(ctx, delivery.SourceBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: no source revision is read until the indexer starts again", delivery.SourceBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err) + default: + each := func(e jetstream.KeyValueEntry) { + if opOf(e) { + applySource(ctx, ix, e.Key(), e.Revision(), e.Value(), logger) + } + } + sw, err := settle(ctx, source, each) + if err != nil { + return err + } + go func() { sourceErrs <- follow(ctx, sw, each) }() + } cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module) if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, forgeStream, err) @@ -289,5 +307,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return nil case err := <-errs: return err + case err := <-sourceErrs: + return err } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 4b8cda5..19b3658 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -67,3 +67,24 @@ func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 12 t.Errorf("terminated %d acked %d keys %d log %q", m.termd, m.acked, len(st.vals), logs.String()) } } + +func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) + var logs bytes.Buffer + logger := log.New(&logs, "", 0) + mods := &memStore{vals: map[string][]byte{}} + ix := &index.Indexer{Store: &memStore{vals: map[string][]byte{}}, Modules: mods, Builds: index.NewBuilds()} + value, err := delivery.EncodeSource(delivery.Source{Group: "g", Heads: map[string]string{"novox/mesh-delivery": head}}) + if err != nil { + t.Fatal(err) + } + applySource(context.Background(), ix, "other", 3, value, logger) // not the key of the mesh's source + applySource(context.Background(), ix, delivery.SourceMainKey, 4, []byte(`{"v":1`), logger) + if !strings.Contains(logs.String(), "source") { + t.Errorf("a source that cannot be read was not said: %q", logs.String()) + } + logs.Reset() + applySource(context.Background(), ix, delivery.SourceMainKey, 5, value, logger) + if logs.Len() != 0 { + t.Errorf("a good source said %q", logs.String()) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go new file mode 100644 index 0000000..48c8513 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go @@ -0,0 +1,27 @@ +package main + +import ( + "context" + "log" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-indexer/internal/index" +) + +// applySource takes one put of the gate's bucket source into the indexer: the key main, at its revision, which is the mesh's +// source revision (mesh-sdk go/delivery SourceMainKey, Source). Any other key is not the mesh's source. A value that is no +// source is said; so is a write that failed, and it is tried again only at the next revision (novox/hq ADR 0437 §8). +func applySource(ctx context.Context, ix *index.Indexer, key string, rev uint64, value []byte, logger *log.Logger) { + if key != delivery.SourceMainKey { + return + } + s, err := delivery.DecodeSource(value) + if err != nil { + logger.Printf("source.%s at revision %d is not a source, and is left: %v", key, rev, err) + return + } + if err := ix.ApplySource(ctx, rev, s); err != nil { + logger.Printf("source.%s at revision %d could not be taken into the modules: %v", key, rev, err) + } +} diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index a03eb51..02011be 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019223929-58373d65b47a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index 9000f11..e7d24d6 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -2,6 +2,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019203925-84a764372337 h1:cZyFnvDi git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019203925-84a764372337/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019223929-58373d65b47a h1:Vjk2Kq+g/Q86N8CCHq3OatJIJ0JSy6Kh980ekGxNLmE= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019223929-58373d65b47a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69 h1:kvtV/SPsdRVUL/BZ3xxGS6U8YrXy/t7TRBjVrD6VLw8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 8bf63af..7be504a 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -6,6 +6,7 @@ import ( "fmt" "slices" "strings" + "sync" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -44,6 +45,9 @@ type Indexer struct { Modules Store Builds *Builds Logf func(format string, args ...any) + + mu sync.Mutex // guards source, which the source watch and the forge messages both reach + source sourceNow } func (ix *Indexer) logf(format string, args ...any) { diff --git a/modules/mesh-delivery-indexer/internal/index/modules.go b/modules/mesh-delivery-indexer/internal/index/modules.go index 47001a6..1243ff5 100644 --- a/modules/mesh-delivery-indexer/internal/index/modules.go +++ b/modules/mesh-delivery-indexer/internal/index/modules.go @@ -45,7 +45,9 @@ func fingerprintOf(k Known, trees map[string]string) (fp, none string) { } // applyMerged writes, for each module built from the merged repository, its fingerprint at the merge commit (or none and -// why). A module of another repository is not written. A value the module already holds is not written again. +// why), with the source revision that names the merge commit as the repository's head when the indexer has seen it +// (ApplySource), and Src 0 when it has not. A module of another repository is not written. A value the module already +// holds is not written again. func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error { for _, k := range ix.Builds.Modules(f.Repository) { key, err := delivery.ModulesKey(k.Module) @@ -54,7 +56,7 @@ func (ix *Indexer) applyMerged(ctx context.Context, f delivery.Forge) error { continue } fp, none := fingerprintOf(k, f.Trees) - next := delivery.Module{Repository: f.Repository, Commit: f.Merge, FP: fp, None: none} + next := delivery.Module{Repository: f.Repository, Commit: f.Merge, FP: fp, None: none, Src: ix.srcOf(f.Repository, f.Merge)} err = write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) { if found { if cur, err := delivery.DecodeModule(value); err == nil && delivery.SameModule(cur, withV(next)) {