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 01e3af36..8bb6284f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go index 89d56bdf..a1afd067 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -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 }} +} diff --git a/modules/mesh-delivery-indexer/internal/index/control_records_test.go b/modules/mesh-delivery-indexer/internal/index/control_records_test.go index 66f7948c..337fb881 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records_test.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records_test.go @@ -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) } diff --git a/modules/mesh-delivery-indexer/internal/index/seatdefs.go b/modules/mesh-delivery-indexer/internal/index/seatdefs.go index 87d3125e..488bb37f 100644 --- a/modules/mesh-delivery-indexer/internal/index/seatdefs.go +++ b/modules/mesh-delivery-indexer/internal/index/seatdefs.go @@ -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. diff --git a/modules/mesh-delivery-indexer/internal/index/startread.go b/modules/mesh-delivery-indexer/internal/index/startread.go new file mode 100644 index 00000000..feea611a --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/startread.go @@ -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 +}