IX19: the indexer reads every contracts module's trunk definitions at its start before judging, waits for the seat git at most two minutes, says view-waits past five and judges again once it answers

Taken from the join base (6ce86c96, ff6e74f9) onto the build branch.

Builds: design 47 G §198, §248, §316, §255, §57 (tests 54, 57)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 11:51:07 +01:00
parent bcd0888928
commit e7bd54e67b
5 changed files with 315 additions and 5 deletions
@@ -591,6 +591,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := ix.Reconcile(ctx); err != nil {
logger.Printf("dropping the modules the trunk does not hold at start: %v", err)
}
// Every contracts module's trunk definitions read before any manifest is judged and before the forge bridge's stream is
// followed (design 47, the indexer reads every definition before it judges): a definition merged while it was down, or
// one no event announced, is never missing from the judging below. Once the state is bound, so on its first binding too.
// Through the seat git (milestone 47/G W4.15, design 47 *A part that cannot read the forge keeps what it has*): a seat that
// does not answer is waited for at most index.GitWaitBound before the judging below runs on the kept definitions, said in
// indexer_first_views and as view-waits past five minutes, and read again on the next forge message heard.
defs := gitWaits(ctx, ix, findings, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views does not say the wait for the seat git", logger.Printf), logger.Printf)
seatDefs.Ready(func(jetstream.KeyValue) { defs.Start(ctx) })
// Every manifest judged at the start, so a judge that changed with this program judges every one again (ADR 0578
// §1), once the verdicts state is there; each merge judges them again (index.ApplyMessage).
verdicts.Ready(func(jetstream.KeyValue) {
@@ -603,13 +611,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}()
// A pass that fails (a read the bus did not answer) is judged again at the next message or source revision the
// indexer hears (index.kickJudge): no timer.
j, err := ix.JudgeAtStart(ctx)
if err != nil {
logger.Printf("judging the manifests at start: %v; judged again at the next message heard", err)
judge := func(fromKept bool) {
j, err := ix.JudgeAtStart(ctx)
if err != nil {
logger.Printf("judging the manifests at start (from kept definitions: %v): %v; judged again at the next message heard", fromKept, err)
return
}
logger.Printf("judged %d manifests at start (from kept definitions: %v): %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v",
j.Judged, fromKept, j.Written, len(j.Refused), j.Outgrown, j.Waiting)
}
if !defs.Waiting() {
judge(false)
return
}
logger.Printf("judged %d manifests at start: %d verdicts written, %d with problems, of which outgrown and kept accepted: %v, waiting for a judge that knows their fields: %v",
j.Judged, j.Written, len(j.Refused), j.Outgrown, j.Waiting)
// The seat git did not answer: the judging waits for it at most index.GitWaitBound, off the start, so nothing else of
// the start is held behind it.
go func() { judge(defs.Ready(ctx)) }()
})
// A build module says each time it wrote its toolchains key: its stream, and the indexer's durable consumer on it, are
// made by the controller as the forge's are. Where the stream is not there yet the indexer says so and is started again
@@ -736,6 +753,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
handle(ctx, m, ix, heard, logger)
ix.KickFill(ctx)
unreadAgain(ctx, ix, logger)
defs.Heard(ctx)
}, logger.Printf)
if err != nil {
return err
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/git"
"sync/atomic"
"time"
@@ -180,3 +181,27 @@ func followAssignments(ctx context.Context, hub *process.EventHub, cred process.
}
return f
}
// gitWaits is the indexer's read of the trunk definitions at its start through the seat git (index.DefinitionsAtStart): its
// wait said as a first view of the seat's (Owner git, State seat-definitions) in the indexer's first views, and its finding
// view-waits written past views.WaitsBound and cleared once the seat answers.
func gitWaits(ctx context.Context, ix *index.Indexer, findings, status *process.AwaitedBucket, logf func(string, ...any)) *index.DefinitionsAtStart {
finding := func(s views.Status, raised bool) {
w := sdkfindings.Writer(ctx, findings.KV)
f := views.ViewWaitsFinding(index.Module, s, time.Now())
var err error
if raised {
err = w.Put(f)
} else {
err = w.Delete(f)
}
if err != nil {
logf("views: the finding view-waits of the seat git is not written: %v", err)
}
}
f := &views.Follower{Owner: git.Seat, Tool: "seat:" + git.Seat + "." + git.VerbTrees, State: index.SeatDefinitionsState, Logf: logf,
RaiseWaits: func(s views.Status) { finding(s, true) }, ClearWaits: func(s views.Status) { finding(s, false) }}
status.Ready(func(kv jetstream.KeyValue) { f.Said = views.KeepStatus(ctx, kv, logf) })
return &index.DefinitionsAtStart{Read: ix.ReadTrunkDefinitions, Waits: f, Logf: logf,
Judge: func(ctx context.Context) error { _, err := ix.JudgeAtStart(ctx); return err }}
}
@@ -125,6 +125,36 @@ func TestTheControlRecords(t *testing.T) {
// and source, says seat-definition-changed only when a digest moved, keeps the trunk's entry over one that does not
// parse (its finding), removes a pull request's entries when it closes, keeps a verdict's out-of-step files, and
// raises generated-types-out-of-step from a trunk build end, cleared only by a clean one of the same module.
// Milestone 47/G W4.15, design 47 test 54: the seat git not answering at start holds the judging at most its bound, which
// then runs on the kept definitions with the wait said; the read once the forge is heard holds the view and judges again.
t.Run("W4.15 the start waits for the seat git, judges from kept definitions, and judges again once it answers", func(t *testing.T) {
answers := []error{fmt.Errorf("mesh-contracts: %w", errForgeRead), fmt.Errorf("again: %w", errForgeRead), nil}
var said []string
judged := 0
d := &DefinitionsAtStart{Bound: time.Millisecond, Waits: waitsFake{&said},
Read: func(context.Context) (int, error) { err := answers[0]; answers = answers[1:]; return 1, err },
Judge: func(context.Context) error { judged++; return nil }}
ctx := context.Background()
d.Start(ctx)
if !d.Waiting() || len(said) != 1 || said[0] != "waits "+GitDefinitionsView {
t.Fatalf("waiting %v, said %q: want the wait for the seat git said once", d.Waiting(), said)
}
if !d.Ready(ctx) || judged != 0 {
t.Fatalf("the start past its bound: want it judged from the kept definitions, judged again %d", judged)
}
d.Heard(ctx) // the forge heard, the seat still not answering: the wait stands, said once
if !d.Waiting() || len(said) != 1 || judged != 0 {
t.Fatalf("a read still not answered: waiting %v, said %q, judged %d", d.Waiting(), said, judged)
}
d.Heard(ctx)
if d.Waiting() || len(said) != 2 || said[1] != "holds "+GitDefinitionsView || judged != 1 || d.Ready(ctx) {
t.Errorf("the seat answering: waiting %v, said %q, judged again %d: want the view held and every manifest judged again once", d.Waiting(), said, judged)
}
d.Heard(ctx) // nothing waits: no read, no judging
if judged != 1 || len(answers) != 0 {
t.Errorf("a forge message with nothing waiting read or judged again")
}
})
t.Run("seat definitions", func(t *testing.T) {
const repo, m1, m2, m3, head = "novox/mesh-contracts", "m1", "m2", "m3", "h1"
manifest := `{"module":"mesh-contracts","version":"1.0","build":{"artifacts":[{"name":"go","kind":"package","language":"go","paths":["go"],"series":"0.1","generated-from":"."}]}}`
@@ -315,6 +345,29 @@ func TestTheControlRecords(t *testing.T) {
t.Errorf("said %+v: only a trunk's definition is said (ADR 0636 §4)", c)
}
}
// Design 47 (the indexer reads every definition before it judges): at its start the indexer reads every contracts
// module's trunk at its repository's head in the source revision, so a definition merged with no event heard is kept
// before any manifest is judged; a repository the source names no head of is read at the module's own commit.
kept, _ = meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "mesh-contracts", Manifest: manifest})
mkey, _ = meshcatalogclient.ManifestsKey(meshcatalogclient.ManifestDigest([]byte(manifest)))
ix.Manifests.(*memStore).vals[mkey] = kept
mod, _ = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: meshcatalogclient.ManifestDigest([]byte(manifest)), Repository: repo, Commit: m3, Path: meshcatalogclient.RootPath})
ix.Modules.(*memStore).vals["mesh-contracts"] = mod
files["h3"] = map[string]string{"seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0500", 1)}
ix.source = sourceNow{rev: 7, heads: map[string]string{repo: "h3"}}
if n, err := ix.ReadTrunkDefinitions(ctx); err != nil || n != 1 {
t.Fatalf("read at start = %d, %v; want the one contracts module", n, err)
}
if e, _ := entry("git.trunk"); e.Commit != "h3" {
t.Errorf("git.trunk after the read at start = %+v, want the source's head h3", e)
}
ix.source = sourceNow{}
if _, err := ix.ReadTrunkDefinitions(ctx); err != nil {
t.Fatal(err)
}
if e, _ := entry("git.trunk"); e.Commit != m3 {
t.Errorf("git.trunk read with no source head = %+v, want the module's own commit", e)
}
// ADR 0636 §5, D-I3, D-I5: a seat of the mesh's own contracts is taken only from its domain's contracts module; a
// move there is a handover in either order of the two trunks; a second definer is refused with a finding naming both.
@@ -636,3 +689,9 @@ func (m memSource) Keys() ([]string, error) {
}
func (m memSource) Get(key string) (*views.Entry, bool, error) { e, ok := m[key]; return e, ok, nil }
// waitsFake records what the indexer says it waits for and holds.
type waitsFake struct{ said *[]string }
func (w waitsFake) WaitFor(_ context.Context, view string) { *w.said = append(*w.said, "waits "+view) }
func (w waitsFake) Holds(view string) { *w.said = append(*w.said, "holds "+view) }
@@ -13,12 +13,14 @@ package index
// contracts module.
import (
"cmp"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"maps"
"path"
"slices"
"strconv"
@@ -30,6 +32,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-delivery/sdk/seatdefs"
"git.novox.be/novox/mesh-delivery/sdk/trunk"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -278,6 +281,49 @@ func (ix *Indexer) SeatDefinitionsAgain(ctx context.Context, module string, pull
return out
}
// ReadTrunkDefinitions reads, at the start and before any manifest is judged, the seat definitions of every contracts
// module the trunk holds, at its repository's head in the source revision (at the module's own commit where the source
// names none), and keeps them as a merge does (design 47, *The indexer reads every definition before it judges*): a
// definition merged while the indexer was down, or one no event announced, is kept before a manifest claiming its seat
// is judged. It answers how many contracts modules were read; one the forge did not answer is in the error, and is read
// again at its next merged message or by indexer_seat_definitions_again.
func (ix *Indexer) ReadTrunkDefinitions(ctx context.Context) (int, error) {
if ix.SeatDefinitions == nil || ix.ReadFile == nil || ix.Modules == nil || ix.Manifests == nil {
return 0, nil
}
held, err := ix.heldModules(ctx)
if err != nil {
return 0, err
}
ix.mu.Lock()
heads := maps.Clone(ix.source.heads)
ix.mu.Unlock()
read := 0
var errs []error
for _, name := range slices.Sorted(maps.Keys(held)) {
m := held[name]
manifest, _, err := ix.manifestRaw(ctx, m.Manifest)
if err != nil {
errs = append(errs, fmt.Errorf("%s's manifest: %w", name, err))
continue
}
dir := cmp.Or(m.Path, meshcatalogclient.RootPath)
commit := m.Commit
for r, head := range heads {
if trunk.SameRepository(r, m.Repository) && head != "" {
commit = head
}
}
for _, c := range contractsModules(map[string]string{dir: manifest}) {
read++
if err := ix.keepTrunk(ctx, m.Repository, commit, c); err != nil {
errs = append(errs, fmt.Errorf("%s at %.12s: %w", name, commit, err))
}
}
}
return read, errors.Join(errs...)
}
// keepTrunk writes a contracts module's trunk definitions at a merge: each that parses where its digest moved (said),
// each the module's index no longer names removed (said with no digest), and the module's unparsed finding written or
// cleared. A definition that does not parse leaves the trunk's entry as it stands.
@@ -0,0 +1,162 @@
package index
import (
"context"
"errors"
"sync"
"time"
)
// GitWaitBound is how long the indexer waits, at its start, for the seat git to answer the read of every contracts module's
// trunk definitions before it judges from the definitions it kept (design 47, *A part that cannot read the forge keeps
// what it has, and says so*): longer than a forge's restart, short enough that a start is not held behind a forge down
// for good.
const GitWaitBound = 2 * time.Minute
// GitDefinitionsView is what the indexer says it waits for in its first views while the seat git does not answer.
const GitDefinitionsView = "trunk seat definitions"
// ViewWaiter is where the indexer says it waits for the seat git and since when, and raises view-waits past its bound
// (mesh-sdk views.Follower's WaitFor and Holds).
type ViewWaiter interface {
WaitFor(ctx context.Context, view string)
Holds(view string)
}
// DefinitionsAtStart is the indexer's read of the trunk definitions at its start through the seat git: the first read,
// a wait of at most Bound for the seat when it does not answer, the judging from the kept definitions after it, and the
// read again once the seat answers, which judges again every manifest judged from kept definitions.
type DefinitionsAtStart struct {
// Read reads every contracts module's trunk definitions (Indexer.ReadTrunkDefinitions).
Read func(ctx context.Context) (int, error)
// Judge judges every manifest again, as at the start (Indexer.JudgeAtStart).
Judge func(ctx context.Context) error
Waits ViewWaiter
Bound time.Duration
Logf func(string, ...any)
mu sync.Mutex
answered chan struct{} // closed once a read is answered
waiting bool // a read was not answered and no read since was
fromKept bool // the judging at start ran on kept definitions
}
func (d *DefinitionsAtStart) init() {
if d.answered == nil {
d.answered = make(chan struct{})
}
}
func (d *DefinitionsAtStart) logf(format string, args ...any) {
if d.Logf != nil {
d.Logf(format, args...)
}
}
// Start reads the trunk definitions once; a seat that does not answer is said as a wait (Waits.WaitFor) and read again on
// the next sign that the forge runs (Heard). A read answered with other errors (a manifest it cannot read) is answered.
func (d *DefinitionsAtStart) Start(ctx context.Context) {
d.mu.Lock()
d.init()
d.mu.Unlock()
d.read(ctx, "at start")
}
func (d *DefinitionsAtStart) read(ctx context.Context, when string) bool {
n, err := d.Read(ctx)
if errors.Is(err, errForgeRead) {
d.mu.Lock()
first := !d.waiting
d.waiting = true
d.mu.Unlock()
if first {
d.logf("the seat git did not answer the read of %d contracts modules' trunk definitions %s: %v; the indexer judges from the definitions it kept after %s, and reads them again once the seat answers", n, when, err, d.bound())
if d.Waits != nil {
d.Waits.WaitFor(ctx, GitDefinitionsView)
}
}
return false
}
if err != nil {
d.logf("reading the seat definitions of %d contracts modules' trunks %s: %v; each is read again at its next merged message or by indexer_seat_definitions_again", n, when, err)
} else {
d.logf("read the seat definitions of %d contracts modules' trunks %s", n, when)
}
d.mu.Lock()
was := d.waiting
d.waiting = false
select {
case <-d.answered:
default:
close(d.answered)
}
again := d.fromKept
d.fromKept = false
d.mu.Unlock()
if was && d.Waits != nil {
d.Waits.Holds(GitDefinitionsView)
}
if again && d.Judge != nil {
if err := d.Judge(ctx); err != nil {
d.logf("judging again the manifests judged from kept definitions: %v; judged again at the next message heard", err)
} else {
d.logf("judged again every manifest judged from kept definitions, now that the seat git answers")
}
}
return true
}
func (d *DefinitionsAtStart) bound() time.Duration {
if d.Bound > 0 {
return d.Bound
}
return GitWaitBound
}
// Ready waits, while the seat git does not answer the read at start, until a read is answered, or Bound passed (no timer: a context's bound), and says whether the
// judging that follows runs on the definitions the indexer kept; it marks it so, and the read that answers later judges
// again (Read's answer).
func (d *DefinitionsAtStart) Ready(ctx context.Context) (fromKept bool) {
d.mu.Lock()
d.init()
answered, waiting := d.answered, d.waiting
d.mu.Unlock()
if !waiting {
return false // read, or not read at all (the seat definitions' state not bound yet): nothing to wait for
}
wait, cancel := context.WithTimeout(ctx, d.bound())
defer cancel()
select {
case <-answered:
return false
case <-wait.Done():
}
d.mu.Lock()
defer d.mu.Unlock()
select {
case <-d.answered:
return false
default:
}
d.fromKept = true
d.logf("the seat git has not answered within %s: the indexer judges from the definitions it kept", d.bound())
return true
}
// Heard is a sign that the forge runs (a message the forge bridge relays): while the indexer waits for the seat git, it
// reads the trunk definitions again.
func (d *DefinitionsAtStart) Heard(ctx context.Context) {
d.mu.Lock()
waiting := d.waiting
d.mu.Unlock()
if waiting {
d.read(ctx, "once the forge was heard")
}
}
// Waiting says the indexer waits for the seat git.
func (d *DefinitionsAtStart) Waiting() bool {
d.mu.Lock()
defer d.mu.Unlock()
return d.waiting
}