diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 4f040bed..3ed9dab2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -493,9 +493,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if kv := kvs[writtenBucket]; kv != nil { // The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow // once the first registrations are written (§95, §353). - p.heard = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) - if err := hearEnds(hub, p.heard); err != nil { - logger.Printf("a pipeline's end is not heard, so no want follows a pipeline until the assigner starts again: %v", err) + p.ends = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) + err := hearEnds(hub, p.ends) + if err == nil && bus != nil { + err = followRegistrations(ctx, hub, bus.Call, p.ends) + } + if err != nil { + logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err) } } hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 69d369af..8c5d6aa2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -58,10 +58,8 @@ type passer struct { // neither is, by node and module, each module on neither list whose condition this process raised (issue 1945). Nil until the // first pass after a start, which clears what a previous process may have raised (startClears). neither map[string]neitherOne - // ends are the pipeline ends the wants follow (pipelines.go); nil is the trunk-only want, kept until the first registrations. + // ends are the pipeline ends the wants follow (pipelines.go), once the first registrations are heard; until then, and nil, the trunk-only want. ends *pipelineEnds - // heard is the same follower while the wants do not follow it yet (ends is nil). - heard *pipelineEnds // acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket). acts *putActs // noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go index 743125a5..5e9ab67e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -5,13 +5,19 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" + "strings" "sync" + "sync/atomic" + "time" "github.com/nats-io/nats.go/jetstream" + mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" mprc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient" mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" @@ -81,6 +87,9 @@ type endStore interface { get(key string) (value []byte, revision uint64, err error) // swap writes value at key if the key stands at revision (0: if it does not stand); errSwapped when it does not. swap(key string, value []byte, revision uint64) error + // keys are the keys that stand, and drop deletes one. + keys() ([]string, error) + drop(key string) error } var errSwapped = errors.New("the key changed") @@ -115,6 +124,16 @@ func (k kvEnds) swap(key string, value []byte, revision uint64) error { return err } +func (k kvEnds) keys() ([]string, error) { + ks, err := k.kv.Keys(k.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return ks, err +} + +func (k kvEnds) drop(key string) error { return k.kv.Delete(k.ctx, key) } + func isWrongSequence(err error) bool { var ae *jetstream.APIError return errors.As(err, &ae) && ae.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence @@ -126,7 +145,9 @@ type pipelineEnds struct { st endStore wake func() logf func(string, ...any) - mu sync.Mutex + // on is whether the wants follow the ends: set by the first registration heard, never before (design 47 G §95, §353). + on atomic.Bool + mu sync.Mutex // firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design // 47 G §14). firsts map[string]string @@ -278,7 +299,7 @@ func writtenTools(st store) []stdio.Tool { // branchesOf is the watched branches the placements of module chose, each once; the default branch when none names one. func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]string) []string { - if e == nil { + if !e.following() { return nil } seen := map[string]bool{} @@ -300,7 +321,7 @@ func (e *pipelineEnds) branchesOf(module string, placed map[string]map[string]st // mayWant says whether module's wants may name fp: a pipeline passed at fp on every branch, or the wants name fp already // (a node placed later, a module taken away). func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[string]mwc.Want) bool { - if e == nil { + if !e.following() { return true } for _, w := range wanted { @@ -318,10 +339,77 @@ func (e *pipelineEnds) mayWant(module, fp string, branches []string, wanted map[ // wrote marks the end the wants of module were written from. func (e *pipelineEnds) wrote(module, fp string, branches []string) { - if e == nil { + if !e.following() { return } for _, b := range branches { e.markWritten(module, b, fp) } } + +// following says whether the wants follow the pipeline ends yet; a nil follower never does. +func (e *pipelineEnds) following() bool { return e != nil && e.on.Load() } + +// registered takes a module's registration (design 47 G §353): the first one heard switches the wants to the ends, its first +// watched branch is the default a placement runs, and the ends kept for a branch it no longer watches are deleted (§352). gone +// is a registration removed: all its ends go. +func (e *pipelineEnds) registered(reg mpcat.Registration, gone bool) { + e.mu.Lock() + if gone || len(reg.Branches) == 0 { + delete(e.firsts, reg.Module) + } else { + e.firsts[reg.Module] = reg.Branches[0] + } + e.mu.Unlock() + kept, err := e.st.keys() + if err != nil { + e.logf("the ends kept cannot be listed: %v", err) + } + prefix := keys.EscapeName(reg.Module) + "." + for _, k := range kept { + rest, ok := strings.CutPrefix(k, prefix) + if !ok { + continue + } + branch, err := keys.UnescapeName(rest) + if err == nil && !gone && slices.Contains(reg.Branches, branch) { + continue + } + if err := e.st.drop(k); err != nil { + e.logf("the end %s is not deleted: %v", k, err) + } + } + if !e.on.Swap(true) { + e.logf("the first registration is heard (%s): the wants follow the pipelines' ends from now", reg.Module) + } + e.wake() +} + +// followRegistrations follows the registrations through the seat mesh-catalog (design 47 G §353): a first view of its verb +// and each registration-changed on the hub. It returns when the hub's handler is registered; the first view is asked on its own +// goroutine, the wants keep the trunk-only form until a registration is heard. +func followRegistrations(ctx context.Context, hub *process.EventHub, call mcc.Call, e *pipelineEnds) error { + regs := mcc.NewRegistrations(call) + regs.Follower.Logf = e.logf + regs.Follower.Changed = func(module string, value json.RawMessage, gone bool, _ time.Time) { + reg := mpcat.Registration{Module: module} + if !gone { + _ = json.Unmarshal(value, ®) + } + e.registered(reg, gone) + } + if err := hub.Handle(mpcat.SubjectRegistrationChanged, func(m jetstream.Msg) { + if err := regs.Hear(ctx, m.Data()); err != nil { + e.logf("%s: %v", m.Subject(), err) + } + _ = m.Ack() + }); err != nil { + return err + } + go func() { + if err := regs.Start(ctx); err != nil && ctx.Err() == nil { + e.logf("the first view of the registrations is not answered: %v", err) + } + }() + return nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go index 17597d49..acb0c207 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go @@ -10,6 +10,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + mpcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" mpr "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" @@ -34,6 +35,13 @@ type memEnds struct { func newMemEnds() *memEnds { return &memEnds{vals: map[string][]byte{}, revs: map[string]uint64{}} } func (m *memEnds) get(key string) ([]byte, uint64, error) { return m.vals[key], m.revs[key], nil } +func (m *memEnds) keys() (out []string, _ error) { + for k := range m.vals { + out = append(out, k) + } + return out, nil +} +func (m *memEnds) drop(key string) error { delete(m.vals, key); delete(m.revs, key); return nil } func (m *memEnds) swap(key string, v []byte, rev uint64) error { if m.revs[key] != rev { return errSwapped @@ -79,6 +87,7 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) { } { woken := 0 e := newPipelineEnds(newMemEnds(), func() { woken++ }, t.Logf) + e.on.Store(true) for _, p := range c.ends { if p.Group != "" { continue // heard and ignored, as hearPipeline does @@ -100,12 +109,14 @@ func TestAWantFollowsAPassedPipeline(t *testing.T) { } // A want that names the fingerprint already is not held by a later failure (a node placed afterwards gets it). e := newPipelineEnds(newMemEnds(), func() {}, t.Logf) + e.on.Store(true) if !e.mayWant("web", fp1, []string{"main"}, map[string]mwc.Want{"g14": {Fp: fp1}}) { t.Error("a want naming the fingerprint was held") } // markWritten keeps the end and says it was written from; a newer end replaces it unwritten. st := newMemEnds() e = newPipelineEnds(st, func() {}, t.Logf) + e.on.Store(true) _ = e.take(pipe("web", "main", 5, 1, fp1, pass), true) e.wrote("web", fp1, []string{"main"}) var end End @@ -153,6 +164,7 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { p.w.put(buckets.Presence, "g14", on) if c.follow { p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf) + p.ends.on.Store(true) if c.passed { _ = p.ends.take(pipe("web", trunkBranch, 5, 1, fp, mpr.PipelineEndedOutcomePassed), true) } @@ -172,3 +184,30 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { } } } + +// Design 47 G §95, §130, §353, §352 (test 100): the wants keep the trunk-only form until the first registration is heard and +// switch on it; a branch a registration no longer watches loses its end, and its first branch is a placement's default. +func TestTheWantsSwitchOnTheFirstRegistration(t *testing.T) { + st := newMemEnds() + woken := 0 + e := newPipelineEnds(st, func() { woken++ }, t.Logf) + _ = e.take(pipe("web", "main", 5, 1, "fp1", mpr.PipelineEndedOutcomeFailed), false) + _ = e.take(pipe("web", "beta", 5, 1, "fp1", mpr.PipelineEndedOutcomePassed), true) + if e.following() || !e.mayWant("web", "fp1", []string{"main"}, nil) { + t.Fatal("the follower held a want before any registration") + } + e.registered(mpcat.Registration{Module: "web", Branches: []string{"beta"}}, false) + if !e.following() || woken == 0 { + t.Error("the first registration did not switch the wants on") + } + if e.mayWant("web", "fp1", []string{"main"}, nil) || !e.mayWant("web", "fp1", []string{"beta"}, nil) { + t.Error("after the switch the wants did not follow the ends") + } + if len(st.vals) != 1 || e.branchOf("web", "") != "beta" || e.branchOf("other", "") != trunkBranch { + t.Errorf("ends kept %d, default branch %q: an unwatched branch's end stayed, or the default is wrong", len(st.vals), e.branchOf("web", "")) + } + e.registered(mpcat.Registration{Module: "web"}, true) + if len(st.vals) != 0 { + t.Errorf("a registration gone left %d ends", len(st.vals)) + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 629dcee0..1db630c4 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -123,11 +123,11 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { + if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed"}) { t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes) } if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", - "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list + "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", "seat:node-engine.manifest-reads", // ADR 0618 §3 diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index d81cca86..d6252329 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -39,6 +39,7 @@ "operator-channel" ], "consumes": [ + "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", @@ -53,6 +54,7 @@ "mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", 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 +}