indexer: a watched branch other than the first is taken in from its source into its own revision, a pure computation from its heads' merged messages

sourceFollow takes source.<branch> for every branch the seat lists (the answer with no branch), ApplyBranchSource computes each module's
fingerprint, inputs and graph node over a view in memory seeded from the branch's newest revision, and keeps it with KeepBranchEntries;
the trunk's modules, manifests, taken state, graph and packages are not written. The trunk branch is named once (index.TrunkBranch).

Builds: design 47 G §13, §14, §291, ADR 0666 §6, A-85, A-90, A-93
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:31:30 +01:00
parent e27160571a
commit b3cd0ef661
8 changed files with 268 additions and 16 deletions
@@ -1414,3 +1414,74 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je
}
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
// TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision: the seat's verb source lists a second watched branch,
// beta, with the merge of mesh-catalog it names as its head (design 47 G §13, §14, §291, A-93); the indexer started with the
// branch listed reads source.beta, takes the head's merged message in as a pure computation, and keeps branch.beta.1, which
// the branch follower (index.BranchView, as the branch's builds read it) reads; the trunk's modules are left.
func TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision(t *testing.T) {
manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
if err != nil {
t.Fatal(err)
}
const merge = "89abcdef0123456789abcdef0123456789abcdef"
data, err := forgemsg.EncodeForge(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "beta", Number: 301,
Head: "0123456789abcdef0123456789abcdef01234567", Merge: merge, Root: "r0",
Trees: map[string]string{"modules/avahi": "betatree"}, Manifests: map[string]string{"modules/avahi": string(manifest)},
Locks: map[string]map[string]string{"modules/avahi": {"go.sum": "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"}}})
if err != nil {
t.Fatal(err)
}
js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) {
builds.end(t, "novox/mesh-controller", "build-agent", "modules/build-agent")
}, func(ctx context.Context, js jetstream.JetStream) {
seedBuildModules(t, ctx, js, nil)
source := func(m *nats.Msg) {
var ask struct{ Branch string }
_ = json.Unmarshal(m.Data, &ask)
answer := meshdelivery.SourceAnswer{Branch: ask.Branch}
switch ask.Branch {
case "":
answer.Branches = []string{"main", "beta"}
case "beta":
answer.Group, answer.Revision = "g", 9
answer.Repositories = []meshdelivery.SourceHead{{Repository: "novox/mesh-catalog", Head: merge, Trunk: "beta"}}
}
raw, _ := json.Marshal(map[string]any{"result": answer, "node": "n1"})
_ = m.Respond(raw)
}
bridge := func(m *nats.Msg) {
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"message": json.RawMessage(data)}, "node": "n1"})
_ = m.Respond(raw)
}
for subject, h := range map[string]nats.MsgHandler{
"mesh.seat.mesh-delivery.call.source.>": source, "mesh.seat.mesh-delivery.tool.source": source,
"mesh.mod.mesh-delivery-forge-bridge.call.bridge_merged_message.>": bridge, "mesh.mod.mesh-delivery-forge-bridge.tool.bridge_merged_message": bridge,
} {
if _, err := js.Conn().Subscribe(subject, h); err != nil {
t.Fatal(err)
}
}
})
objects := newBusObjects(js, testAccount, index.CatalogRevisionsObjects)
var view index.Store
var at int64
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && at == 0; time.Sleep(50 * time.Millisecond) {
if v, _, n, err := index.BranchView(ctx, objects, "beta"); err == nil {
view, at = v, n
}
}
if at != 1 {
t.Fatalf("the branch follower reads beta at revision %d within 20 s of the start, want 1", at)
}
value, _, found, err := view.Get(ctx, "avahi")
if err != nil || !found {
t.Fatalf("branch.beta.1 holds no avahi (%v)", err)
}
if m, err := meshcatalogclient.DecodeModule(value); err != nil || m.Commit != merge || m.Inputs2 == nil {
t.Errorf("avahi at the branch's head = %+v (%v), want commit %s with its inputs", m, err, merge)
}
if e, err := modules.Get(ctx, "avahi"); err == nil {
t.Errorf("modules.avahi = %s, want the trunk's modules left", e.Value())
}
}
@@ -511,7 +511,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The source revision of a watched branch (design 47 G §15, §115, §362): asked of the seat mesh-delivery's verb source as a
// kept call, once at the start, before the handler of its event source-moved reads anything; a source-moved heard
// meanwhile is read after.
source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, logf: logger.Printf}
source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, applyBranch: ix.ApplyBranchSource, logf: logger.Printf}
go source.catchUp(ctx)
// Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the
// assignments is settled first, so the toolchains settle through the gate already knowing it.
@@ -191,6 +191,9 @@ func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt(
s = &sourceFollow{
read: func(branch string) (meshdelivery.SourceAnswer, error) {
order = append(order, "read "+branch)
if branch == "" { // given no branch the seat lists the branches it keeps a source of (A-93)
return meshdelivery.SourceAnswer{Branches: []string{"main", "beta", "feat=2e1"}}, nil
}
if revision == 0 {
return meshdelivery.SourceAnswer{}, fmt.Errorf("source.main is not readable")
}
@@ -205,15 +208,21 @@ func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt(
order = append(order, fmt.Sprintf("apply %s %d", a.Branch, a.Revision))
return nil
},
applyBranch: func(_ context.Context, branch string, a meshdelivery.SourceAnswer) error {
order = append(order, fmt.Sprintf("branch %s %d", branch, a.Revision))
return nil
},
logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) },
}
s.catchUp(context.Background())
want := []string{"read main", "apply main 4", "read main", "apply main 4"} // once, after the catch-up; beta is left
// The trunk, then each other watched branch the seat lists (a name unescaped), then what was heard meanwhile once each,
// beta among them: main through the take-in, every other branch into its own revision (A-93).
want := []string{"read main", "apply main 4", "read ", "read beta", "branch beta 4", "read feat=2e1", "branch feat.1 4", "read main", "apply main 4", "read beta", "branch beta 4"}
if !reflect.DeepEqual(order, want) {
t.Errorf("order = %v, want %v", order, want)
}
s.moved(context.Background(), "main") // after the catch-up: read at once
if len(order) != 6 {
if len(order) != 13 {
t.Errorf("a source-moved after the catch-up was not read at once: %v", order)
}
order, revision = nil, 0
@@ -237,7 +237,7 @@ func readVia(call toolCaller) func(ctx context.Context, repository, sha, dir str
}
// trunkBranch is the branch pull requests are open into.
const trunkBranch = "main"
const trunkBranch = index.TrunkBranch
// openVia asks the forge which pull requests into the trunk are open now (the seat git's open-pulls), every repository's.
// An answer that lacks the list is refused, never read as none.
@@ -8,6 +8,7 @@ import (
"sync"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/keys"
"github.com/nats-io/nats.go/jetstream"
)
@@ -40,17 +41,31 @@ func sourceVia(call toolCaller) func(branch string) (meshdelivery.SourceAnswer,
type sourceFollow struct {
read func(branch string) (meshdelivery.SourceAnswer, error)
apply func(context.Context, meshdelivery.SourceAnswer) error
logf func(string, ...any)
// applyBranch takes a watched branch other than the trunk's in (design 47 G §291, A-93); branch is its name, unescaped.
applyBranch func(ctx context.Context, branch string, a meshdelivery.SourceAnswer) error
logf func(string, ...any)
mu sync.Mutex
caught bool
waiting map[string]bool
}
// catchUp reads the trunk branch's source revision and applies it, then the branches a source-moved named meanwhile. A
// read that fails is said; the catch-up ends all the same, and the next source-moved reads again.
// catchUp reads the trunk branch's source revision and applies it, then every other branch the seat keeps a source of (the
// answer given no branch lists them, A-93), then the branches a source-moved named meanwhile. A read that fails is said;
// the catch-up ends all the same, and the next source-moved reads again.
func (s *sourceFollow) catchUp(ctx context.Context) {
s.take(ctx, trunkBranch)
var others []string
if list, err := s.read(""); err != nil {
s.logf("the branches the source names are not read, and each is read at its next source-moved: %v", err)
} else {
others = list.Branches
}
for _, branch := range others {
if branch != trunkBranch {
s.take(ctx, branch)
}
}
s.mu.Lock()
s.caught = true
branches := s.waiting
@@ -76,17 +91,28 @@ func (s *sourceFollow) moved(ctx context.Context, branch string) {
s.take(ctx, branch)
}
// take reads one branch's source revision and applies it. A branch other than the trunk's is left until the indexer keeps
// a catalog graph per watched branch.
// take reads one branch's source revision (branch as the key name source.<branch> has it) and applies it: the trunk's
// through the take-in, another watched branch's into its own revision (A-93).
func (s *sourceFollow) take(ctx context.Context, branch string) {
if branch != trunkBranch {
return
}
a, err := s.read(branch)
if err != nil {
s.logf("source.%s is not read, and is read again at its next source-moved: %v", branch, err)
return
}
if branch != trunkBranch {
name, err := keys.UnescapeName(branch)
if err != nil {
s.logf("source.%s names no branch: %v", branch, err)
return
}
if s.applyBranch == nil {
return
}
if err := s.applyBranch(ctx, name, a); err != nil {
s.logf("source.%s at revision %d could not be taken into the branch's revision: %v", branch, a.Revision, err)
}
return
}
if err := s.apply(ctx, a); err != nil {
s.logf("source.%s at revision %d could not be taken into the modules: %v", branch, a.Revision, err)
}
@@ -0,0 +1,55 @@
package index
import (
"context"
"errors"
"maps"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/process"
)
// ApplyBranchSource takes a watched branch other than the first in from the source revision the seat answered for it
// (design 47 G §13, §14, §291, ADR 0666 §6, A-85, A-93): a pure computation from the merged messages of the heads the source
// names, over a view in memory seeded from the branch's newest revision, written straight into the branch's next revision
// (KeepBranchEntries): each module's fingerprint, inputs and graph node at the branch's heads, computed as the trunk's are
// (applyMoving). Nothing of the trunk is written: its modules, manifests, taken state, graph and packages stay its own,
// and the branch has no bucket of its own. A head whose merged message is not read keeps no revision, said; the next
// source-moved of the branch takes it in again.
func (ix *Indexer) ApplyBranchSource(ctx context.Context, branch string, a meshdelivery.SourceAnswer) error {
if isTrunkBranch(branch) || ix.Revisions == nil || ix.Message == nil || len(a.Repositories) == 0 {
return nil
}
heads := map[string]string{}
for _, r := range a.Repositories {
heads[r.Repository] = r.Head
}
view, nodes, _, err := BranchView(ctx, ix.Revisions, branch)
if errors.Is(err, process.ErrNotYet) {
ix.logf("catalog graph: the branch %s is not taken in yet: its objects %s are not on the bus", branch, CatalogRevisionsObjects)
return nil
}
if err != nil {
return err
}
d := ix.dry()
// The branch's own view: its modules and graph from its revision, no taken merge, and the packages it publishes in memory
// over the trunk's, never written to them.
d.ix.Modules, d.ix.Graph, d.ix.Taken = view, nodes, newOverlay(nil)
if ix.Packages != nil {
d.ix.Packages = newOverlay(ix.Packages)
}
d.ix.source = sourceNow{rev: uint64(a.Revision), heads: maps.Clone(heads)}
d.ix.Logf = func(format string, args ...any) {
ix.logf("branch "+branch+", computed and not written: "+format, args...)
}
if unread := d.ix.takeInHeads(ctx, heads); len(unread) > 0 {
ix.logf("catalog graph: the branch %s at source revision %d keeps no revision: %d repositories were not taken in (%v); taken in again at its next source-moved", branch, a.Revision, len(unread), unread)
return nil
}
entries, err := d.ix.viewEntries(ctx)
if err != nil {
return err
}
return ix.KeepBranchEntries(ctx, branch, a.Revision, entries)
}
@@ -0,0 +1,75 @@
package index
import (
"context"
"encoding/json"
"fmt"
"maps"
"slices"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
)
// Design 47 G §13, §14, §291, A-93: a watched branch other than the first is taken in as a pure computation from the merged
// messages of its heads, straight into branch.<branch>.<n>; the trunk's modules, graph and revision stay the trunk's.
func TestABranchOtherThanTheFirstIsTakenInIntoItsOwnRevisionAndTheTrunkIsLeftAlone(t *testing.T) {
ctx := context.Background()
ix, mods, merged := newFormIndexerUnmerged(t, builderModules(t)...)
ix.Revisions = newMemObjects(0)
ix.Graph = newMem()
ix.Message = func(_ context.Context, repository, head string, _ int) ([]byte, error) {
if repository != catalogueRepo || head != merged.Merge {
return nil, fmt.Errorf("no merge of %s at %s", repository, head)
}
return json.Marshal(merged)
}
var said []meshcatalog.CatalogRevisionEvent
ix.SayCatalogRevision = func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error {
said = append(said, e)
return nil
}
before := map[string]string{}
keys, _ := mods.Keys(ctx)
for _, k := range keys {
v, _, _, _ := mods.Get(ctx, k)
before[k] = string(v)
}
graphBefore, _ := ix.Graph.Keys(ctx)
answer := meshdelivery.SourceAnswer{Branch: "beta", Revision: 9, Repositories: []meshdelivery.SourceHead{{Repository: catalogueRepo, Head: merged.Merge, Trunk: "beta"}}}
if err := ix.ApplyBranchSource(ctx, "beta", answer); err != nil {
t.Fatal(err)
}
if len(said) != 1 || said[0].Revision != 1 || said[0].Branch != "beta" || said[0].Source != 9 || said[0].Entries != int64(len(builderModules(t))) {
t.Fatalf("the branch take-in said %+v, want revision 1 of beta at source 9 with every module", said)
}
rev, entries, err := ReadRevision(ctx, ix.Revisions, BranchObject("beta", 1))
if err != nil || rev.Branch != "beta" || len(entries) != len(builderModules(t)) {
t.Fatalf("branch.beta.1 = %+v with %d entries (%v)", rev, len(entries), err)
}
for _, e := range entries {
if e.Commit != merged.Merge || e.Inputs == nil { // computed at the branch's head, with the inputs its fingerprint stands on
t.Errorf("the entry of %s is at %q with inputs %v, want the branch's head %s with inputs", e.Module, e.Commit, e.Inputs, merged.Merge)
}
}
if _, any, _ := newestTrunk(ctx, ix.Revisions); any {
t.Error("a branch take-in kept a trunk revision")
}
after := map[string]string{}
keys, _ = mods.Keys(ctx)
for _, k := range keys {
v, _, _, _ := mods.Get(ctx, k)
after[k] = string(v)
}
if !maps.Equal(before, after) {
t.Errorf("a branch take-in wrote the trunk's modules")
}
if graphAfter, _ := ix.Graph.Keys(ctx); !slices.Equal(graphBefore, graphAfter) {
t.Errorf("a branch take-in wrote the trunk's graph: %v, was %v", graphAfter, graphBefore)
}
// Taken in again at the same heads, nothing moved: no second revision.
if err := ix.ApplyBranchSource(ctx, "beta", answer); err != nil || len(said) != 1 {
t.Errorf("the same heads said %d revisions (%v), want one", len(said), err)
}
}
@@ -246,15 +246,25 @@ func (ix *Indexer) KeepView(ctx context.Context) error {
if ix.Revisions == nil || ix.Modules == nil || ix.catchingUp.Load() {
return nil
}
held, err := ix.heldModules(ctx)
entries, err := ix.viewEntries(ctx)
if err != nil {
return err
}
return ix.keepEntries(ctx, entries)
}
// viewEntries is the entry of every module the indexer's view (its modules and its module graph) holds: the module's value
// with its node of the graph, as KeepView keeps them.
func (ix *Indexer) viewEntries(ctx context.Context) (map[string]meshcatalog.CatalogEntry, error) {
held, err := ix.heldModules(ctx)
if err != nil {
return nil, err
}
nodes := graph.Graph{}
if ix.Graph != nil {
keys, err := ix.Graph.Keys(ctx)
if err != nil && !errors.Is(err, process.ErrNotYet) {
return fmt.Errorf("the module graph: %w", err)
return nil, fmt.Errorf("the module graph: %w", err)
}
for _, k := range keys {
if value, _, found, err := ix.Graph.Get(ctx, k); err == nil && found {
@@ -275,7 +285,7 @@ func (ix *Indexer) KeepView(ctx context.Context) error {
}
entries[name] = ix.withArtifacts(ctx, CatalogEntryOf(name, cur, m, raw, nodes[name]), m)
}
return ix.keepEntries(ctx, entries)
return entries, nil
}
// keepEntries keeps entries as the trunk revision after the newest, says catalog-revision, and keeps none when no entry
@@ -411,8 +421,14 @@ func (ix *Indexer) RoomAgain(ctx context.Context) (bool, error) {
// graph (NodeOf), and the revision it was read at (0 when the bucket keeps none yet: both start empty and the first start's
// catch-up fills them). A lost entry is an error naming the module (ReadRevision), never a module left out of the view.
func TrunkView(ctx context.Context, o Objects) (modules, nodes Store, at int64, err error) {
return BranchView(ctx, o, TrunkBranch)
}
// BranchView is TrunkView of a watched branch: the view read from the newest revision the bucket keeps of that branch
// (design 47 G §291, A-93), empty and at 0 when it keeps none.
func BranchView(ctx context.Context, o Objects, branch string) (modules, nodes Store, at int64, err error) {
view, held := newOverlay(nil), newOverlay(nil)
newest, any, err := newestTrunk(ctx, o)
newest, any, err := newestOf(ctx, o, branch)
if err != nil || !any {
return view, held, 0, err
}