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

Stamp the modules with the gate's source revision, read from source.main (issue 1344)

Browse Source
This commit is contained in:
mesh-tools
2026-10-20 00:30:52 +00:00
parent 5ba1da5b82
commit 409c41d4ae
7 changed files with 82 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+23 -3
View File
@@ -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
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go
+21
View File
@@ -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())
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go
+27
View File
@@ -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)
}
}
modules/mesh-delivery-indexer/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-delivery-indexer/go.sum
+2
View File
@@ -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=
modules/mesh-delivery-indexer/internal/index/candidates.go
+4
View File
@@ -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) {
modules/mesh-delivery-indexer/internal/index/modules.go
+4 -2
View File
@@ -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)) {