Write each module's source fingerprint at a trunk merge in the indexer (issues 1239, 1209)
This commit is contained in:
@@ -40,84 +40,22 @@ accounts {
|
||||
// manifest's state (mesh-delivery-indexer_candidates, history 10) and mesh-controller_builds holding build records, and
|
||||
// no mesh-delivery-gate_source bucket.
|
||||
func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
conf := filepath.Join(dir, "bus.conf")
|
||||
if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
srv, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go srv.Start()
|
||||
if !srv.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(srv.Shutdown)
|
||||
|
||||
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(ctl.Close)
|
||||
js, err := jetstream.New(ctl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, "FORGE", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
candidates, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.CandidatesBucket, History: delivery.CandidatesHistory})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ModulesBucket, History: delivery.ModulesHistory}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
builds, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.BuildsBucket, History: 5})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Build records in the controller's layout (mesh-sdk go/delivery/testdata/build-record-build-agent.json, captured from
|
||||
// the controller's encoder), under delivery.BuildKey, with the newest pointer beside them.
|
||||
for _, b := range []struct{ repo, module, path, id string }{
|
||||
{"novox/mesh-delivery", "a", "modules/a", "build-1"}, {"novox/mesh-catalog", "c", "modules/c", "build-2"},
|
||||
} {
|
||||
rec, _ := json.Marshal(map[string]any{"id": b.id, "state": "built", "repository": b.repo, "ref": "main", "module": b.module,
|
||||
"commit": "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", "path": b.path, "at": "2026-10-19T09:38:00Z"})
|
||||
if _, err := builds.Put(ctx, delivery.BuildKey(b.module, b.id), rec); err != nil {
|
||||
t.Fatal(err)
|
||||
js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds jetstream.KeyValue) {
|
||||
// Build records in the controller's layout (mesh-sdk go/delivery/testdata/build-record-build-agent.json, captured from
|
||||
// the controller's encoder), under delivery.BuildKey, with the newest pointer beside them.
|
||||
for _, b := range []struct{ repo, module, path, id string }{
|
||||
{"novox/mesh-delivery", "a", "modules/a", "build-1"}, {"novox/mesh-catalog", "c", "modules/c", "build-2"},
|
||||
} {
|
||||
rec, _ := json.Marshal(map[string]any{"id": b.id, "state": "built", "repository": b.repo, "ref": "main", "module": b.module,
|
||||
"commit": "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", "path": b.path, "at": "2026-10-19T09:38:00Z"})
|
||||
if _, err := builds.Put(ctx, delivery.BuildKey(b.module, b.id), rec); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := builds.Put(ctx, delivery.NewestKey(b.module), []byte(b.id)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := builds.Put(ctx, delivery.NewestKey(b.module), []byte(b.id)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
broker := filepath.Join(t.TempDir(), "broker")
|
||||
cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"})
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
|
||||
})
|
||||
h1, h2 := "0123456789abcdef0123456789abcdef01234567", "89abcdef0123456789abcdef0123456789abcdef"
|
||||
for _, f := range []delivery.Forge{
|
||||
{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: h1, Action: delivery.PullUpdated, Dirs: []string{"modules/a"}, DirsKnown: true},
|
||||
@@ -154,3 +92,114 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// startIndexer starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill
|
||||
// mesh-controller_builds, and runs the program. mesh-delivery-gate_source is never created.
|
||||
func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) {
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
conf := filepath.Join(dir, "bus.conf")
|
||||
if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
srv, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go srv.Start()
|
||||
if !srv.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(srv.Shutdown)
|
||||
|
||||
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(ctl.Close)
|
||||
js, err := jetstream.New(ctl)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, "FORGE", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
candidates, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.CandidatesBucket, History: delivery.CandidatesHistory})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
modules, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ModulesBucket, History: delivery.ModulesHistory})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
builds, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.BuildsBucket, History: 5})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
put(ctx, builds)
|
||||
|
||||
broker := filepath.Join(t.TempDir(), "broker")
|
||||
cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"})
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
|
||||
return js, ctx, candidates, modules
|
||||
}
|
||||
|
||||
// TestIndexerWritesModulesAtAMerge: a module's record and its newest pointer are in mesh-controller_builds; a merged
|
||||
// message on FORGE makes modules.<m> within 5 s, the fingerprint of the merge's tree and the recorded inputs.
|
||||
func TestIndexerWritesModulesAtAMerge(t *testing.T) { // 1239/AC9
|
||||
// Inputs and record in the controller's layout: SourceInputs under its Go field names (mesh-sdk
|
||||
// go/delivery/testdata/build-record-build-agent.json, captured from the controller's encoder).
|
||||
const inputs = `{"Module":"a","Tree":"oldtree","Bases":["registry.example/base@sha256:aaa"],"Contexts":null,"Toolchains":{"code":"go sha256:ccc 0123456789abcdef"},"Unpinned":""}`
|
||||
js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds jetstream.KeyValue) {
|
||||
rec := `{"id":"build-1","state":"built","repository":"novox/mesh-delivery","ref":"main","module":"a","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"modules/a","at":"2026-10-19T09:38:00Z","source-inputs":` + inputs + `}`
|
||||
if _, err := builds.Put(ctx, delivery.BuildKey("a", "build-1"), []byte(rec)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := builds.Put(ctx, delivery.NewestKey("a"), []byte("build-1")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
const merge = "89abcdef0123456789abcdef0123456789abcdef"
|
||||
data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12,
|
||||
Head: "0123456789abcdef0123456789abcdef01234567", Merge: merge, Root: "r0", Trees: map[string]string{"modules/a": "newtree"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged", data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want, _ := delivery.SourceFingerprint(delivery.SourceInputs{Module: "a", Tree: "newtree", Bases: []string{"registry.example/base@sha256:aaa"},
|
||||
Toolchains: map[string]string{"code": "go sha256:ccc 0123456789abcdef"}})
|
||||
var m delivery.Module
|
||||
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if e, err := modules.Get(ctx, "a"); err == nil {
|
||||
if m, err = delivery.DecodeModule(e.Value()); err == nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
if m.FP != want || m.Repository != "novox/mesh-delivery" || m.Commit != merge {
|
||||
t.Errorf("module = %+v within 5 s, want fp %s", m, want)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -238,6 +238,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
modules, err := bucket(delivery.ModulesBucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
builds, err := bucket(delivery.BuildsBucket)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -260,7 +264,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
errs := make(chan error, 1)
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
|
||||
ix := &index.Indexer{Store: kvStore{candidates}, Builds: known, Logf: logger.Printf}
|
||||
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Builds: known, Logf: logger.Printf}
|
||||
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)
|
||||
|
||||
@@ -4,6 +4,7 @@ package index
|
||||
import (
|
||||
"encoding/json"
|
||||
"path"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
@@ -17,6 +18,14 @@ type Builds struct {
|
||||
mu sync.Mutex
|
||||
records map[string]delivery.BuildRecord // by bucket key
|
||||
byDir map[string]delivery.BuildRecord // by repository and path; nil when stale
|
||||
mods map[string]*moduleBuilds // by module; nil when stale
|
||||
}
|
||||
|
||||
// moduleBuilds is what the builds say of one module: its newest record of any kind, and its newest successful trunk
|
||||
// build that recorded its inputs.
|
||||
type moduleBuilds struct {
|
||||
any delivery.BuildRecord
|
||||
trunk *delivery.BuildRecord
|
||||
}
|
||||
|
||||
// NewBuilds is an empty Builds.
|
||||
@@ -35,7 +44,7 @@ func (b *Builds) Put(key string, value []byte) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.records[key] = r
|
||||
b.byDir = nil
|
||||
b.byDir, b.mods = nil, nil
|
||||
}
|
||||
|
||||
// Delete takes a key out.
|
||||
@@ -44,7 +53,7 @@ func (b *Builds) Delete(key string) {
|
||||
defer b.mu.Unlock()
|
||||
if _, ok := b.records[key]; ok {
|
||||
delete(b.records, key)
|
||||
b.byDir = nil
|
||||
b.byDir, b.mods = nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -75,3 +84,53 @@ func (b *Builds) ModuleOf(repository, dir string) string {
|
||||
}
|
||||
return path.Base(strings.Trim(dir, "/"))
|
||||
}
|
||||
|
||||
// Known is a module the builds know of in a repository: its directory there and its newest successful trunk build (ref
|
||||
// main, state built, with source inputs) if any. The newest key of a module is any state and ref, so it is not what
|
||||
// finds this (mesh-sdk go/delivery/buildrecord.go NewestKey); the records are.
|
||||
type Known struct {
|
||||
Module string
|
||||
Repository string
|
||||
Path string
|
||||
Trunk *delivery.BuildRecord
|
||||
}
|
||||
|
||||
func newer(a, b delivery.BuildRecord) bool {
|
||||
return a.At.After(b.At) || (a.At.Equal(b.At) && a.ID > b.ID)
|
||||
}
|
||||
|
||||
// Modules are the modules built from a repository, by name. A module whose newest successful trunk build is of the
|
||||
// repository belongs to it with that record's path; a module with no such build belongs to the repository of its newest
|
||||
// record of any kind.
|
||||
func (b *Builds) Modules(repository string) []Known {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.mods == nil {
|
||||
b.mods = map[string]*moduleBuilds{}
|
||||
for _, r := range b.records {
|
||||
m := b.mods[r.Module]
|
||||
if m == nil {
|
||||
m = &moduleBuilds{any: r}
|
||||
b.mods[r.Module] = m
|
||||
} else if newer(r, m.any) {
|
||||
m.any = r
|
||||
}
|
||||
if r.Ref == "main" && r.State == "built" && r.SourceInputs != nil && (m.trunk == nil || newer(r, *m.trunk)) {
|
||||
c := r
|
||||
m.trunk = &c
|
||||
}
|
||||
}
|
||||
}
|
||||
var out []Known
|
||||
for name, m := range b.mods {
|
||||
k := Known{Module: name, Repository: m.any.Repository, Path: strings.Trim(m.any.Path, "/"), Trunk: m.trunk}
|
||||
if m.trunk != nil {
|
||||
k.Repository, k.Path = m.trunk.Repository, strings.Trim(m.trunk.Path, "/")
|
||||
}
|
||||
if k.Repository == repository {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -37,9 +37,11 @@ const writeTries = 8
|
||||
|
||||
// Indexer writes candidates.<group> from the forge bridge's messages.
|
||||
type Indexer struct {
|
||||
Store Store
|
||||
Builds *Builds
|
||||
Logf func(format string, args ...any)
|
||||
Store Store // candidates
|
||||
// Modules is the modules bucket; nil writes no fingerprints.
|
||||
Modules Store
|
||||
Builds *Builds
|
||||
Logf func(format string, args ...any)
|
||||
}
|
||||
|
||||
func (ix *Indexer) logf(format string, args ...any) {
|
||||
@@ -50,11 +52,17 @@ func (ix *Indexer) logf(format string, args ...any) {
|
||||
|
||||
// Apply takes one message of the bridge's stream FORGE into the group of its branch. A pull message adds or moves the
|
||||
// pull request, a closed one or a merged message takes it out; a group left with none is deleted. A change is written
|
||||
// by compare-and-set on the revision read, and read again on a conflict. A lone pull request is a group of one.
|
||||
// by compare-and-set on the revision read, and read again on a conflict. A lone pull request is a group of one. A
|
||||
// merged message also writes the fingerprint of each module of its repository (modules.go).
|
||||
func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error {
|
||||
if f.Kind != delivery.ForgePull && f.Kind != delivery.ForgeMerged {
|
||||
return nil
|
||||
}
|
||||
if f.Kind == delivery.ForgeMerged && ix.Modules != nil {
|
||||
if err := ix.applyMerged(ctx, f); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if trunkLike[f.Branch] {
|
||||
ix.logf("%s#%d: its branch %q is no delivery group, so no candidate", f.Repository, f.Number, f.Branch)
|
||||
return nil
|
||||
@@ -63,45 +71,58 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
|
||||
}
|
||||
for try := 0; try < writeTries; try++ {
|
||||
value, revision, found, err := ix.Store.Get(ctx, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading candidates.%s: %w", key, err)
|
||||
}
|
||||
return write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
var cur *delivery.Candidate
|
||||
if found {
|
||||
c, err := delivery.DecodeCandidate(value)
|
||||
if err != nil {
|
||||
return fmt.Errorf("candidates.%s holds what is not a candidate: %w", key, err)
|
||||
return nil, false, fmt.Errorf("candidates.%s holds what is not a candidate: %w", key, err)
|
||||
}
|
||||
cur = &c
|
||||
}
|
||||
next := ix.next(cur, f)
|
||||
switch {
|
||||
case cur == nil && next == nil, cur != nil && next != nil && delivery.SameCandidate(*cur, *next):
|
||||
return nil
|
||||
case next == nil && cur == nil, next != nil && cur != nil && delivery.SameCandidate(*cur, *next):
|
||||
return nil, false, nil
|
||||
case next == nil:
|
||||
err = ix.Store.Delete(ctx, key, revision)
|
||||
case cur == nil:
|
||||
var data []byte
|
||||
if data, err = delivery.EncodeCandidate(*next); err == nil {
|
||||
err = ix.Store.Create(ctx, key, data)
|
||||
}
|
||||
return nil, true, nil
|
||||
}
|
||||
data, err := delivery.EncodeCandidate(*next)
|
||||
return data, false, err
|
||||
})
|
||||
}
|
||||
|
||||
// write changes one key of a store by compare-and-set: compute is given the value read (found says there is one) and
|
||||
// answers the new value, or del to remove the key, or neither for no change. A conflict reads again, up to writeTries.
|
||||
func write(ctx context.Context, st Store, key string, compute func(value []byte, found bool) (next []byte, del bool, err error)) error {
|
||||
for try := 0; try < writeTries; try++ {
|
||||
value, revision, found, err := st.Get(ctx, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading %s: %w", key, err)
|
||||
}
|
||||
next, del, err := compute(value, found)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch {
|
||||
case next == nil && !del:
|
||||
return nil
|
||||
case del:
|
||||
err = st.Delete(ctx, key, revision)
|
||||
case !found:
|
||||
err = st.Create(ctx, key, next)
|
||||
default:
|
||||
var data []byte
|
||||
if data, err = delivery.EncodeCandidate(*next); err == nil {
|
||||
err = ix.Store.Update(ctx, key, data, revision)
|
||||
}
|
||||
err = st.Update(ctx, key, next, revision)
|
||||
}
|
||||
if errors.Is(err, ErrConflict) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("writing candidates.%s: %w", key, err)
|
||||
return fmt.Errorf("writing %s: %w", key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("candidates.%s changed under every one of %d tries", key, writeTries)
|
||||
return fmt.Errorf("%s changed under every one of %d tries", key, writeTries)
|
||||
}
|
||||
|
||||
// next is the group after the message; nil when the group is left with no pull request.
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The reasons a module has no fingerprint (ADR 0437 §5; issue 1209's decision).
|
||||
const (
|
||||
// NoneNarrow is for a module whose image compiles a Go program from a context: the context is the hash of the build
|
||||
// source the builder handed it, not a git tree.
|
||||
NoneNarrow = "an artifact compiles a Go program from a build source that git trees do not pin: no fingerprint before release 4"
|
||||
// NoneContext is for a context that is a repository's root tree.
|
||||
NoneContext = "an artifact reads the root tree of a repository the indexer does not know the tree of before release 4"
|
||||
// NoneNoBuild is for a module no successful trunk build recorded the inputs of.
|
||||
NoneNoBuild = "no successful trunk build recorded its inputs"
|
||||
)
|
||||
|
||||
// fingerprintOf is the module's fingerprint at the merge, or none and why: its directory's tree from the merged
|
||||
// message and the bases, contexts and toolchains its newest successful trunk build recorded.
|
||||
func fingerprintOf(k Known, trees map[string]string) (fp, none string) {
|
||||
if k.Trunk == nil {
|
||||
return "", NoneNoBuild
|
||||
}
|
||||
in := *k.Trunk.SourceInputs
|
||||
artifacts := make([]string, 0, len(in.Contexts))
|
||||
for a := range in.Contexts {
|
||||
artifacts = append(artifacts, a)
|
||||
}
|
||||
sort.Strings(artifacts)
|
||||
for _, a := range artifacts {
|
||||
if strings.HasPrefix(in.Contexts[a], "narrow:") {
|
||||
return "", NoneNarrow
|
||||
}
|
||||
}
|
||||
if len(artifacts) > 0 {
|
||||
return "", NoneContext
|
||||
}
|
||||
in.Tree = trees[k.Path] // the directory "" is the repository's root
|
||||
return delivery.SourceFingerprint(in)
|
||||
}
|
||||
|
||||
// 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.
|
||||
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)
|
||||
if err != nil {
|
||||
ix.logf("%s: the module %q has no usable key, skipped: %v", f.Repository, k.Module, err)
|
||||
continue
|
||||
}
|
||||
fp, none := fingerprintOf(k, f.Trees)
|
||||
next := delivery.Module{Repository: f.Repository, Commit: f.Merge, FP: fp, None: none}
|
||||
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)) {
|
||||
return nil, false, nil
|
||||
}
|
||||
}
|
||||
data, err := delivery.EncodeModule(next)
|
||||
return data, false, err
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("modules.%s: %w", key, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func withV(m delivery.Module) delivery.Module { m.V = delivery.Version; return m }
|
||||
@@ -0,0 +1,203 @@
|
||||
package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
const merge = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
|
||||
// record is a build record as the controller writes it (mesh-sdk go/delivery/testdata/build-record-build-agent.json is
|
||||
// the captured shape): SourceInputs under its Go field names, no json keys. inputs is that object's JSON.
|
||||
func record(t *testing.T, id, module, repo, path, ref, state, at, inputs string) (string, []byte) {
|
||||
t.Helper()
|
||||
raw := `{"id":"` + id + `","state":"` + state + `","repository":"` + repo + `","ref":"` + ref + `","module":"` + module + `","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"` + path + `","at":"` + at + `"`
|
||||
if inputs != "" {
|
||||
raw += `,"source-fingerprint":"src1:ignored","source-inputs":` + inputs
|
||||
}
|
||||
raw += `}`
|
||||
var check delivery.BuildRecord
|
||||
if err := json.Unmarshal([]byte(raw), &check); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return delivery.BuildKey(module, id), []byte(raw)
|
||||
}
|
||||
|
||||
func modulesIndexer(t *testing.T, recs ...[2]any) (*Indexer, *memStore) {
|
||||
t.Helper()
|
||||
mods := newMem()
|
||||
ix := &Indexer{Store: newMem(), Modules: mods, Builds: NewBuilds()}
|
||||
for _, r := range recs {
|
||||
ix.Builds.Put(r[0].(string), r[1].([]byte))
|
||||
}
|
||||
return ix, mods
|
||||
}
|
||||
|
||||
func rec(t *testing.T, id, module, repo, path, ref, state, at, inputs string) [2]any {
|
||||
k, v := record(t, id, module, repo, path, ref, state, at, inputs)
|
||||
return [2]any{k, v}
|
||||
}
|
||||
|
||||
func mergedMsg(repo string, trees map[string]string) delivery.Forge {
|
||||
return delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: repo, Branch: "feat/x", Number: 12, Head: h1, Merge: merge, Root: "r0", Trees: trees}
|
||||
}
|
||||
|
||||
func moduleOf(t *testing.T, st *memStore, key string) (delivery.Module, bool) {
|
||||
t.Helper()
|
||||
v, ok := st.vals[key]
|
||||
if !ok {
|
||||
return delivery.Module{}, false
|
||||
}
|
||||
m, err := delivery.DecodeModule(v)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return m, true
|
||||
}
|
||||
|
||||
const aInputs = `{"Module":"a","Tree":"oldtree","Bases":["registry.example/base@sha256:aaa"],"Contexts":null,"Toolchains":{"code":"go sha256:ccc 0123456789abcdef"},"Unpinned":""}`
|
||||
|
||||
func TestModuleFingerprintFromTheMergeTreeAndTheRecordedInputs(t *testing.T) { // 1239/AC1
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", aInputs))
|
||||
if err := ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree", "modules/b": "other"})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want, none := delivery.SourceFingerprint(delivery.SourceInputs{Module: "a", Tree: "newtree", Bases: []string{"registry.example/base@sha256:aaa"},
|
||||
Toolchains: map[string]string{"code": "go sha256:ccc 0123456789abcdef"}})
|
||||
if none != "" {
|
||||
t.Fatal(none)
|
||||
}
|
||||
m, ok := moduleOf(t, mods, "a")
|
||||
if !ok || m.FP != want || m.None != "" || m.Repository != "novox/mesh-delivery" || m.Commit != merge || m.Src != 0 {
|
||||
t.Errorf("module = %+v (%v), want fp %s", m, ok, want)
|
||||
}
|
||||
if _, ok := mods.vals["b"]; ok {
|
||||
t.Error("a key was written for a directory no build names")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeOfTheRecordedTreeReproducesTheRecordedFingerprint(t *testing.T) { // 1239/AC2
|
||||
// mailu, build-1792391134620131370: mesh-catalog at b4527d9091e972f28cb365623d9280045e7741a3, dir modules/mailu. Its
|
||||
// inputs and recorded fingerprint are the live ones mesh-sdk go/delivery/fingerprint_test.go TestFingerprintLiveBuilds
|
||||
// captured (its tree is `git rev-parse <commit>:modules/mailu`, its bases from the build's [bases] log line).
|
||||
const tree = "cfc6fa8ff5fc1e220132e51ade653b9d3f387b3e"
|
||||
const recorded = "src1:539f2e64000d436c572ef53b40e524012455945f8c5a1c6bb173ee0707e88ef4"
|
||||
inputs := `{"Module":"mailu","Tree":"` + tree + `","Bases":["sha256:25f3cfeaceca14921366af4d1240b56457ef46273bdb508c7b0e8f469f6fd228"],"Contexts":null,` +
|
||||
`"Toolchains":{"code":"typescript sha256:4b81682df5665c1f461d3f720a3a0fd7593f4aa71e4fc272dbfdc4bbd235b489 20125bd590041690"},"Unpinned":""}`
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1792391134620131370", "mailu", "novox/mesh-catalog", "modules/mailu", "main", "built", "2026-10-19T09:00:00Z", inputs))
|
||||
if err := ix.Apply(context.Background(), mergedMsg("novox/mesh-catalog", map[string]string{"modules/mailu": tree})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m, ok := moduleOf(t, mods, "mailu"); !ok || m.FP != recorded {
|
||||
t.Errorf("module = %+v (%v), want fp %s", m, ok, recorded)
|
||||
}
|
||||
}
|
||||
|
||||
func TestANarrowContextHasNoFingerprint(t *testing.T) { // 1239/AC3, 1209/AC1
|
||||
// A Go program's build source as the builder records it (build-agent's record, mesh-sdk go/delivery/testdata):
|
||||
// the context "server" is "narrow:" and a hash.
|
||||
inputs := `{"Module":"build-agent","Tree":"11c0b4f90cae0d6e4ef9cc48cbfdf9f6dbe52bca","Bases":["sha256:d9e853e87e55526f6b2917df91a2115c36dd7c696a35be12163d44e6e2a4b6bc"],"Contexts":{"server":"narrow:d51ca744fd11af59d616c858f76c27b3e58e16e11fbdc0c720eb9d7d19d55fac"},"Toolchains":null,"Unpinned":""}`
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "build-agent", "novox/mesh-controller", "modules/build-agent", "main", "built", "2026-10-19T09:00:00Z", inputs))
|
||||
if err := ix.Apply(context.Background(), mergedMsg("novox/mesh-controller", map[string]string{"modules/build-agent": "newtree"})); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, ok := moduleOf(t, mods, "build-agent")
|
||||
if !ok || m.FP != "" || !strings.Contains(m.None, "Go program") || !strings.Contains(m.None, "release 4") {
|
||||
t.Errorf("module = %+v (%v), want none naming the Go program's build source and release 4", m, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnpinnedBuildHasNoFingerprint(t *testing.T) { // 1239/AC4
|
||||
inputs := `{"Module":"a","Tree":"t","Bases":null,"Contexts":null,"Toolchains":null,"Unpinned":"the build resolves packages when built"}`
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", inputs))
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
|
||||
if m, ok := moduleOf(t, mods, "a"); !ok || m.FP != "" || m.None != "the build resolves packages when built" {
|
||||
t.Errorf("module = %+v (%v)", m, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestARootTreeContextHasNoFingerprint(t *testing.T) { // 1239/AC5
|
||||
inputs := `{"Module":"a","Tree":"t","Bases":null,"Contexts":{"web":"f00dcafef00dcafef00dcafef00dcafef00dcafe"},"Toolchains":null,"Unpinned":""}`
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", inputs))
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
|
||||
m, ok := moduleOf(t, mods, "a")
|
||||
if !ok || m.FP != "" || !strings.Contains(m.None, "root tree") || !strings.Contains(m.None, "release 4") {
|
||||
t.Errorf("module = %+v (%v)", m, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDirectoryGoneFromTheMergeHasNoTree(t *testing.T) { // 1239/AC6
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", aInputs))
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/other": "t"}))
|
||||
if m, ok := moduleOf(t, mods, "a"); !ok || m.FP != "" || m.None != "no tree" {
|
||||
t.Errorf("module = %+v (%v)", m, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAModuleWithNoSuccessfulTrunkBuildHasNone(t *testing.T) { // 1239/AC7
|
||||
ix, mods := modulesIndexer(t,
|
||||
rec(t, "build-1", "failed-only", "novox/mesh-delivery", "modules/f", "main", "failed", "2026-10-19T09:00:00Z", aInputs),
|
||||
rec(t, "build-2", "branch-only", "novox/mesh-delivery", "modules/g", "feat/x", "built", "2026-10-19T09:00:00Z", aInputs),
|
||||
rec(t, "build-3", "no-inputs", "novox/mesh-delivery", "modules/h", "main", "built", "2026-10-19T09:00:00Z", ""),
|
||||
)
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/f": "t", "modules/g": "t", "modules/h": "t"}))
|
||||
for _, name := range []string{"failed-only", "branch-only", "no-inputs"} {
|
||||
if m, ok := moduleOf(t, mods, name); !ok || m.FP != "" || !strings.Contains(m.None, "no successful trunk build") {
|
||||
t.Errorf("%s: module = %+v (%v)", name, m, ok)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheNewestSuccessfulTrunkBuildCounts(t *testing.T) { // 1239 (the newest key is only a hint)
|
||||
older := strings.Replace(aInputs, `"Bases":["registry.example/base@sha256:aaa"]`, `"Bases":["registry.example/base@sha256:old"]`, 1)
|
||||
ix, mods := modulesIndexer(t,
|
||||
rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", older),
|
||||
rec(t, "build-2", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T10:00:00Z", aInputs),
|
||||
rec(t, "build-3", "a", "novox/mesh-delivery", "modules/a", "main", "failed", "2026-10-19T11:00:00Z", older),
|
||||
rec(t, "build-4", "a", "novox/mesh-delivery", "modules/a", "feat/x", "built", "2026-10-19T12:00:00Z", older),
|
||||
)
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
|
||||
want, _ := delivery.SourceFingerprint(delivery.SourceInputs{Module: "a", Tree: "newtree", Bases: []string{"registry.example/base@sha256:aaa"},
|
||||
Toolchains: map[string]string{"code": "go sha256:ccc 0123456789abcdef"}})
|
||||
if m, _ := moduleOf(t, mods, "a"); m.FP != want {
|
||||
t.Errorf("fp = %s, want the 10:00 trunk build's %s", m.FP, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestModulesOfOtherRepositoriesAreNotWritten(t *testing.T) { // 1239/AC8
|
||||
ix, mods := modulesIndexer(t,
|
||||
rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", aInputs),
|
||||
rec(t, "build-2", "c", "novox/mesh-catalog", "modules/c", "main", "built", "2026-10-19T09:00:00Z", strings.Replace(aInputs, `"Module":"a"`, `"Module":"c"`, 1)))
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "t", "modules/c": "t"}))
|
||||
if _, ok := mods.vals["c"]; ok {
|
||||
t.Error("the key of a module of another repository was written")
|
||||
}
|
||||
if _, ok := mods.vals["a"]; !ok {
|
||||
t.Error("the key of the merged repository's module was not written")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnUnchangedModuleWritesNothing(t *testing.T) { // 1239 (unchanged value writes nothing)
|
||||
ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", aInputs))
|
||||
m := mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"})
|
||||
_ = ix.Apply(context.Background(), m)
|
||||
rev, writes := mods.revs["a"], mods.writes
|
||||
_ = ix.Apply(context.Background(), m)
|
||||
if mods.revs["a"] != rev || mods.writes != writes {
|
||||
t.Errorf("the same merge wrote again (%d writes, was %d)", mods.writes, writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergedMessageRemovesThePullFromItsGroupToo(t *testing.T) { // 1238/AC5 with 1239 (both effects of one message)
|
||||
ix, _ := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "built", "2026-10-19T09:00:00Z", aInputs))
|
||||
cands := ix.Store.(*memStore)
|
||||
_ = ix.Apply(context.Background(), pull("novox/mesh-delivery", 12, "feat/x", h1, "modules/a"))
|
||||
_ = ix.Apply(context.Background(), mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
|
||||
if _, ok := cands.vals["feat/x"]; ok {
|
||||
t.Error("the merged pull request is still a candidate")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user