diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go index ecd4431e..9a36fcf2 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go @@ -5,7 +5,6 @@ import ( "fmt" "time" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" @@ -179,7 +178,7 @@ func outOfStepTool(state func(string) views.Source) stdio.Tool { "build of the same module), then each open pull request whose check's verdict named out-of-step files.", Input: map[string]any{"type": "object", "properties": map[string]any{}}, Run: func(map[string]any) (any, error) { - rows, err := index.OutOfStepOf(state(index.SeatDefinitionsState), state(mcc.FindingsState)) + rows, err := index.OutOfStepOf(state(index.SeatDefinitionsState), state(index.FindingsState)) if err != nil { return nil, err } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index baa19b47..ef756cfb 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -477,6 +477,9 @@ func waitViews(t *testing.T, ctx context.Context, js jetstream.JetStream, state, const views0Stream = "EVENTS" +// findingsStream is the bus name of the indexer's stream of its findings' events. +const findingsStream = "INDEXER_FINDINGS" + // acceptedHeard counts what a waiter on manifest-accepted heard. var acceptedHeard atomic.Int64 @@ -640,6 +643,12 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before . if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { t.Fatal(err) } + // The indexer's own stream of its findings' events (design 47 G §383), which its finding writer says into (A-30). + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: findingsStream, Subjects: []string{ + "mesh.stream." + index.Module + ".event.finding-written.*", "mesh.stream." + index.Module + ".event.finding-cleared.*", + "mesh.stream." + index.Module + ".event.findings-stated.*"}}); err != nil { + t.Fatal(err) + } put(ctx, buildLog{ctx: ctx, js: js}) for _, b := range before { b(ctx, js) 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 cb57a985..54eb0e52 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -31,8 +31,10 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" stdio "git.novox.be/novox/mesh-sdk/go" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" ) @@ -401,6 +403,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro unjudged := process.AwaitBucket(ctx, js, conn, index.UnjudgedBucket, "a take-in's new manifests are kept for their judging in its memory alone", logger.Printf) tryRecovery := process.AwaitBucket(ctx, js, conn, index.TryRecoveryBucket, "how the recovery of the open tries went is not kept", logger.Printf) findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf) + // The indexer's findings are the seat mesh-conditions' generated finding, written through mesh-sdk's held finding writer + // (design 47 G §689, A-29): it holds what is written before the state is bound, and starts once the state is. + writer, err := views.NewHeldWriter(views.WriterConfig[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent]{ + Module: cred.Module, State: calls.State(index.FindingsState), Node: cred.Node, Asked: mcseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logger.Printf}) + if err != nil { + return fmt.Errorf("the indexer's finding writer cannot be made: %w", err) + } + defer writer.Close() + findings.Ready(func(jetstream.KeyValue) { + // A writer that cannot start (the stream of its asks is not there yet) is said; its findings are then kept in the + // state alone and said by the next start. + if err := writer.Start(ctx); err != nil { + logger.Printf("the finding writer cannot start, its findings are kept in the state alone: %v", err) + } + }) // The seat definitions (design 26, DLV-I17): a state this delivery brings, bound when it appears; until then none is kept. seatDefs := process.AwaitBucket(ctx, js, conn, index.SeatDefinitionsBucket, "no seat definition is kept", logger.Printf) marks := process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf) @@ -413,7 +430,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }) readWhenBound(seatReader, seatDefs, marks) - ix := &index.Indexer{Findings: awaitedStore{b: findings}, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, + ix := &index.Indexer{Findings: writer, OldSections: awaitedStore{b: oldSections}, Unjudged: awaitedStore{b: unjudged}, Requirements: awaitedStore{b: requirements}, Store: kvStore{candidates}, Modules: modules, Manifests: kvStore{manifests}, Taken: awaitedStore{b: taken}, Builds: known, Verdicts: awaitedStore{b: verdicts}, SayRefused: sayRefused(conn, cred, logger.Printf), SayAccepted: sayAccepted(conn, cred), Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), @@ -492,11 +509,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The indexer's one pull of its durable on the shared events stream (process.EventHub): every handler is registered // before it starts. hub := process.NewEventHub(logger.Printf) - followAssignments(ctx, hub, cred, calls.Call, gate, func(ctx context.Context) { + reader := &views.ProcessReader{Module: index.Module, Call: calls.Call, Logf: logger.Printf, Hub: hub, + Findings: func() views.WireWriter { return writer }} + followAssignments(ctx, reader, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views reads nothing and indexer_first_view_again is not heard", logger.Printf), cred, js, gate, func(ctx context.Context) { if err := ix.Recompute(ctx); err != nil { logger.Printf("writing the modules again after the assignments of a build module changed: %v", err) } - }, findings, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "indexer_first_views reads nothing and indexer_first_view_again is not heard", logger.Printf), logger.Printf) + }, logger.Printf) tools := newToolchainWatch(ctx, js, toolchainSink{ put: gate.Toolchain, apply: func(ctx context.Context, module, key string, value []byte, deleted bool) error { @@ -565,7 +584,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // 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) + defs := gitWaits(ctx, ix, writer, cred.Node, 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). @@ -601,7 +620,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // with the build module, as with the gate's source. // Each consumer below that is not there yet, or cannot be heard, is the finding family-not-heard naming its stream until // it binds at a later start (heard.Writer; ADR 0615 §5), so the indexer is not healthy while what it waits on is lost. - notHeard := notHeardWriter(ctx, findings, logger.Printf) + notHeard := notHeardWriter(writer, logger.Printf) // A consumer that is there and cannot be heard stops the indexer, as before; it is said as the finding first. var stopped error hear := func(stream, consequence string, each func(jetstream.Msg)) { @@ -706,8 +725,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro heard := awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} // Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was // said up to; a first view (the tools bundle) answers at that mark. - startViews(ctx, js, cred, marks, - findings, + startViews(ctx, js, cred, marks, writer, map[string]readyOf{"manifests": boundNow(manifests), "candidates": boundNow(candidates), index.VerdictsState: awaited(verdicts), index.PredefinedState: awaited(predefined), index.ConfigCandidatesState: awaited(configCandidates), index.PackagesState: awaited(packages), diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 7496f17a..dd54b4e6 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -17,6 +17,7 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" ) @@ -80,7 +81,7 @@ var testAccount = process.Account{Node: "n1", Module: "mesh-delivery-indexer"} func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) // Review round 2: the indexer's views emitter writes its finding view-unsaid (Emitter.Findings set). - if em := viewsEmitter(context.Background(), nil, process.Account{}, nil, &process.AwaitedBucket{}, nil, t.Logf); em.Findings == nil { + if em := viewsEmitter(nil, process.Account{}, nil, noWire{}, t.Logf); em.Findings == nil { t.Error("the indexer's views emitter writes no view-unsaid finding") } // Milestone 47/G: the object buckets are the module's own by the name its account runs as, never a fixed name. @@ -245,25 +246,5 @@ func TestTheSeatsSourceAnswerIsReadAndItsErrorIsAnError(t *testing.T) { // desig func ptr[T any](v T) *T { return &v } -// findingsKV is a findings bucket of the two calls the indexer's call finding makes. -type findingsKV struct { - jetstream.KeyValue - vals map[string][]byte -} - -func (f *findingsKV) Put(_ context.Context, key string, value []byte) (uint64, error) { - f.vals[key] = value - return uint64(len(f.vals)), nil -} - -func (f *findingsKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { - if _, ok := f.vals[key]; !ok { - return nil, jetstream.ErrKeyNotFound - } - return nil, nil -} - -func (f *findingsKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { - delete(f.vals, key) - return nil -} +// noWire is a finding writer that is none: a non-nil WireWriter for the test of which emitter is made. +type noWire struct{ views.WireWriter } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go index 5372a403..9061d7c6 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/packages_integration_test.go @@ -332,6 +332,20 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T) if rows := outOfStep(2); rows[0].Module != "mesh-contracts" || !strings.Contains(rows[0].Summary, "types.go") || rows[1].Pull != "novox/mesh-contracts#82" { t.Errorf("indexer_out_of_step = %+v, want the trunk's finding and the pull request's files", rows) } + // The finding was said by the indexer's finding writer on its own stream (design 47 G §689, A-30): finding-written. + findingWritten := func() bool { + st, err := js.Stream(ctx, findingsStream) + if err != nil { + return false + } + info, err := st.Info(ctx) + return err == nil && info.State.Msgs > 0 + } + for deadline := time.Now().Add(20 * time.Second); !findingWritten() && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + } + if !findingWritten() { + t.Error("the out-of-step finding was written into the findings state but its finding-written was not said on the indexer's findings stream") + } logEnd(builds.BuildDone) if rows := outOfStep(1); rows[0].Source != index.SourcePull { t.Errorf("after a clean build end indexer_out_of_step = %+v, want the pull request's files alone", rows) 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 a1afd067..3bba0087 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -1,20 +1,23 @@ package main import ( + "cmp" "context" "encoding/json" + "fmt" "git.novox.be/novox/mesh-delivery/contracts/go/seats/git" + "strings" "sync/atomic" "time" "github.com/nats-io/nats.go/jetstream" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/process" - sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/views" @@ -23,7 +26,7 @@ import ( // findingsBucket is the indexer's own findings, which the condition keeper raises and clears (ADR 0564 §2): a value too // large for its views event, and its first view of the assigner's assignments unanswered past its bound. -var findingsBucket = mcc.FindingsBucket(index.Module) +var findingsBucket = index.FindingsBucket // readyOf binds a state's bucket: now for one bound at the start, once it appears for one awaited. type readyOf func(func(jetstream.KeyValue)) @@ -33,19 +36,11 @@ func boundNow(kv jetstream.KeyValue) readyOf { return func(f func(jetstream.KeyV func awaited(b *process.AwaitedBucket) readyOf { return b.Ready } // startViews says each change of the indexer's viewed states on its views family (ADR 0615 §5), once its marks state -// is bound: each state from where its mark says, as soon as its own bucket is bound. A watch that ends is sent on errs. -func startViews(ctx context.Context, js jetstream.JetStream, cred process.Account, marks, findings *process.AwaitedBucket, states map[string]readyOf, errs chan<- error, logf func(string, ...any)) { - tooLarge := func(state, key string, size int) { - kv := findings.KV() - if kv == nil { - return - } - if err := sdkfindings.Writer(ctx, findings.KV).Put(views.TooLargeFinding(index.Module, state, key, size, time.Now())); err != nil { - logf("views: the finding of %s.%s too large is not written: %v", state, key, err) - } - } +// is bound: each state from where its mark says, as soon as its own bucket is bound. A watch that ends is sent on errs. Its +// findings (a value too large, a change not said) go through the indexer's finding writer (design 47 G §687). +func startViews(ctx context.Context, js jetstream.JetStream, cred process.Account, marks *process.AwaitedBucket, writer views.WireWriter, states map[string]readyOf, errs chan<- error, logf func(string, ...any)) { marks.Ready(func(mk jetstream.KeyValue) { - em := viewsEmitter(ctx, js, cred, mk, findings, tooLarge, logf) + em := viewsEmitter(js, cred, mk, writer, logf) for _, state := range sortedStates(states) { states[state](func(kv jetstream.KeyValue) { if err := em.Start(ctx, state, kv, errs); err != nil { @@ -57,11 +52,10 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun } // viewsEmitter is the indexer's emitter of its views family: a change not said past views.Bound is its finding -// view-unsaid, in its own findings once bound (review round 2). -func viewsEmitter(ctx context.Context, js jetstream.JetStream, cred process.Account, mk jetstream.KeyValue, findings *process.AwaitedBucket, - tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter { +// view-unsaid, written through its finding writer (review round 2). +func viewsEmitter(js jetstream.JetStream, cred process.Account, mk jetstream.KeyValue, writer views.WireWriter, logf func(string, ...any)) views.Emitter { return views.Emitter{Owner: index.Module, Subject: cred.EventSubject, Publish: views.JetStreamPublish(js), - Marks: mk, TooLarge: tooLarge, Findings: sdkfindings.Writer(ctx, findings.KV), Logf: logf} + Marks: mk, Findings: writer, Logf: logf} } func sortedStates(states map[string]readyOf) []string { @@ -79,30 +73,36 @@ func sortedStates(states map[string]readyOf) []string { return out } -// notHeardWriter writes the indexer's finding of a family it does not hear into its findings state, once that is bound. -func notHeardWriter(ctx context.Context, findings *process.AwaitedBucket, logf func(string, ...any)) heard.Writer { +// notHeardWriter writes the indexer's finding of a family it does not hear through its finding writer (design 47 G §689). +func notHeardWriter(w index.FindingWriter, logf func(string, ...any)) heard.Writer { return heard.Writer{Observer: index.Module, Logf: logf, - Put: func(key string, value []byte) error { - findings.Ready(func(kv jetstream.KeyValue) { - if _, err := kv.Put(ctx, key, value); err != nil { - logf("the finding %s is not written: %v", key, err) - } - }) - return nil + Put: func(_ string, value []byte) error { + var f mcseat.Finding + if err := json.Unmarshal(value, &f); err != nil { + return err + } + return w.Write(context.Background(), f) }, Delete: func(key string) error { - findings.Ready(func(kv jetstream.KeyValue) { - if _, err := kv.Get(ctx, key); err != nil { - return // none stands - } - if err := kv.Delete(ctx, key); err != nil { - logf("the finding %s is not deleted: %v", key, err) - } - }) - return nil + scope, rest, _ := strings.Cut(key, ".") + i := strings.LastIndex(rest, ".") + if i < 0 { + return fmt.Errorf("%q is no finding's key", key) + } + return clearHeld(context.Background(), w, scope, rest[:i], rest[i+1:], "the indexer hears it now") }} } +// clearHeld clears the finding of scope, id and kind when the writer holds it: none standing says nothing. +func clearHeld(ctx context.Context, w index.FindingWriter, scope, id, kind, why string) error { + for _, f := range w.Held() { + if f.Scope == scope && f.ID == id && f.Kind == kind { + return w.Clear(ctx, scope, id, kind, why) + } + } + return nil +} + // graphState is the module graph's state, named apart from run's bucket of the same name. const graphState = graph.State @@ -114,46 +114,38 @@ var firstViewsBucket = index.Module + "_" + views.StatusState const assignmentsTool = index.AssignerModule + "." + index.AssignmentsState // followAssignments reads the seat mesh-module-assigner's assignments through the assigner's first view and then its views -// family, never its bucket (ADR 0615 §1, DLV-F1r), into gate: each key as it changes, settled once answered, with -// changed told whenever what the gate passes on changed. It waits for the first answer at most views.Bound, so a -// normal start, the two owners coming up together, computes the fingerprints once; past it the indexer goes on, the gate -// passing every report, and raises its condition naming the assigner and assignments until the answer clears it. -func followAssignments(ctx context.Context, hub *process.EventHub, cred process.Account, call func(string, any) (json.RawMessage, error), - gate *buildrun.ToolchainGate, changed func(context.Context), findings, status *process.AwaitedBucket, logf func(string, ...any)) *views.Follower { - finding := func(s views.Status, raised bool) { - w := sdkfindings.Writer(ctx, findings.KV) - f := views.UnansweredFinding(index.Module, s, time.Now()) - var err error - if raised { - err = w.Put(f) - } else { - err = w.Delete(f) +// family, never its bucket (ADR 0615 §1, DLV-F1r), into gate: the module's views reader follows it (design 47 G §532, A-26), +// which asks the first view, hears the views family and raises and clears the finding of a first view unanswered through the +// indexer's finding writer. Each key is taken as it changes, settled once answered, with changed told whenever what the gate +// passes on changed. It waits for the first answer at most views.Bound, so a normal start, the two owners coming up together, +// computes the fingerprints once; past it the indexer goes on, the gate passing every report, and raises its finding naming the +// assigner and assignments until the answer clears it. +func followAssignments(ctx context.Context, reader *views.ProcessReader, status *process.AwaitedBucket, cred process.Account, js jetstream.JetStream, + gate *buildrun.ToolchainGate, changed func(context.Context), logf func(string, ...any)) { + var settled atomic.Bool // the first view answered: the gate's nodes are the counted ones from now on + pic, err := reader.Follow(index.AssignerModule, index.AssignmentsState, assignmentsTool) + if err != nil { + logf("the assignments of the assigner is not followed: %v", err) + return + } + f := pic.Follower() + keep := f.Changed // the picture's own, kept + f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) { + if keep != nil { + keep(key, value, deleted, at) } - if err != nil { - logf("views: the finding of the first view %s is not written: %v", s.Tool, err) + if gate.Assignments(key, value, deleted, at) { + changed(ctx) + } + if settled.Load() && gate.OnPlaced != nil { + gate.OnPlaced() // the counted nodes may have changed (ADR 0624 §1) } } - var settled atomic.Bool // the first view answered: the gate's nodes are the counted ones from now on - f := &views.Follower{Owner: index.AssignerModule, Tool: assignmentsTool, State: index.AssignmentsState, - Ask: views.AskVia(call, assignmentsTool), Logf: logf, - Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) }, - Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) { - if gate.Assignments(key, value, deleted, at) { - changed(ctx) - } - if settled.Load() && gate.OnPlaced != nil { - gate.OnPlaced() // the counted nodes may have changed (ADR 0624 §1) - } - }} - status.Ready(func(kv jetstream.KeyValue) { - f.Said = views.KeepStatus(ctx, kv, logf) - if err := views.HearAgain(ctx, kv, []*views.Follower{f}, logf); err != nil { - logf("views: indexer_first_view_again is not heard (%v) until the indexer starts again", err) - } - }) answered := make(chan struct{}) go func() { - if err := f.Start(ctx); err != nil { + select { + case <-f.Ready(): + case <-ctx.Done(): return } if gate.Settled() { @@ -165,13 +157,7 @@ func followAssignments(ctx context.Context, hub *process.EventHub, cred process. } close(answered) }() - // The assigner's views family, on the module's one hub (the host starts it once every handler is registered); a family the - // hub cannot register is the finding family-not-heard until the indexer starts again (ADR 0615 §5). - // The hub says when the bus made its consumer: a picture asked before then is asked again (views.Follower.ConsumerMade). - hub.OnBound(func(created time.Time) { f.ConsumerMade(ctx, created) }) - _ = notHeardWriter(ctx, findings, logf).Bind(process.EventsStream, "the assignments is the first view's alone, asked again by hand, until the indexer starts again", func() error { - return hub.Handle("mesh.mod."+index.AssignerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, map[string]*views.Follower{index.AssignmentsState: f}, logf)) - }) + reader.Start(ctx, js, cred, status) wait, cancel := context.WithTimeout(ctx, views.Bound) // a context's bound, no timer (ADR 0437 §8) defer cancel() select { @@ -179,21 +165,24 @@ func followAssignments(ctx context.Context, hub *process.EventHub, cred process. case <-wait.Done(): logf("the assigner has not answered assignments within %s: the indexer goes on, every node's toolchain report counting until it answers", views.Bound) } - 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 { +// view-waits written past views.WaitsBound and cleared once the seat answers, by hand through the finding writer (design 47 G A-28). +func gitWaits(ctx context.Context, ix *index.Indexer, w index.FindingWriter, node string, 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()) + id := keys.EscapeName(index.Module + "." + s.Owner + "." + s.State) var err error if raised { - err = w.Put(f) + at := time.Now().UTC() + summary := fmt.Sprintf("%s waits for the view %s.%s at revision %s since %s and acts on nothing new until it holds it; it holds %s", + index.Module, s.Owner, s.State, s.WaitsFor, s.WaitingSince.UTC().Format(time.RFC3339), cmp.Or(s.View, "none")) + err = w.Write(ctx, mcseat.Finding{Scope: views.FindingScope, ID: id, Kind: views.ViewWaitsKind, Severity: mcseat.FindingSeverityWarning, + Summary: summary, Headline: index.Module + " waits for a view of " + s.Owner, Source: "views", Observer: index.Module, At: at, + Subject: mcseat.Subject{Scope: views.FindingScope, ID: index.Module, Machine: node}}) } else { - err = w.Delete(f) + err = clearHeld(ctx, w, views.FindingScope, id, views.ViewWaitsKind, "the part no longer observes it") } if err != nil { logf("views: the finding view-waits of the seat git is not written: %v", err) diff --git a/modules/mesh-delivery-indexer/internal/index/accepted.go b/modules/mesh-delivery-indexer/internal/index/accepted.go index 586ab4f6..0180466b 100644 --- a/modules/mesh-delivery-indexer/internal/index/accepted.go +++ b/modules/mesh-delivery-indexer/internal/index/accepted.go @@ -60,31 +60,24 @@ func (ix *Indexer) sayAccepted(ctx context.Context, a meshcatalog.ManifestAccept return err } -// unsaidKey is the key of module's unsaid finding. -func unsaidKey(module string) string { - return mcc.Key(mcc.ScopeModule, Module+".accepted."+module, UnsaidKind) -} +// unsaidID is the id of module's unsaid finding. +func unsaidID(module string) string { return Module + ".accepted." + module } -// unsaid writes module's finding for err, or deletes it with none. +// unsaid raises module's finding for err, or clears it with none. func (ix *Indexer) unsaid(ctx context.Context, module string, err error) { if ix.Findings == nil { return } - key := unsaidKey(module) - werr := write(ctx, ix.Findings, key, func(value []byte, found bool) ([]byte, bool, error) { - if err == nil { - return nil, found, nil - } - at := ix.now().UTC() - summary := fmt.Sprintf("%s of %s was not published: %v", meshcatalog.EventManifestAccepted, module, err) - data, ferr := mcc.EncodeFinding(mcc.Observation{Scope: mcc.ScopeModule, ID: Module + ".accepted." + module, Kind: UnsaidKind, - Summary: summary, - Headline: "The indexer could not tell the bus module that " + module + "'s manifest was accepted", - Explanation: "The bus module creates and removes a module's bus state from this event. The indexer says it again at its " + - "next judging pass and clears this once it is said; indexer_say_accepted says it now.", - Resolved: "manifest-accepted of " + module + " was said.", Source: UnsaidKind}.Finding(Module, "", at)) - return data, false, ferr - }) + var werr error + if err == nil { + werr = ix.clearFinding(ctx, mcc.ScopeModule, unsaidID(module), UnsaidKind, "manifest-accepted of "+module+" was said.") + } else { + werr = ix.Findings.Write(ctx, ix.observed(mcc.ScopeModule, unsaidID(module), UnsaidKind, + fmt.Sprintf("%s of %s was not published: %v", meshcatalog.EventManifestAccepted, module, err), + "The indexer could not tell the bus module that "+module+"'s manifest was accepted", + "The bus module creates and removes a module's bus state from this event. The indexer says it again at its "+ + "next judging pass and clears this once it is said; indexer_say_accepted says it now.", UnsaidKind)) + } if werr != nil { ix.logf("judge: the finding of %s for %s is not written: %v", meshcatalog.EventManifestAccepted, module, werr) } @@ -149,25 +142,14 @@ func (ix *Indexer) sayFor(ctx context.Context, module string, allowLeft, again b return fmt.Sprintf("said for %s, manifest %s at revision %d, %d bus-state entries", module, a.Digest, a.Revision, len(a.BusState)), nil } -// SayUnsaid says again each manifest-accepted whose finding stands (the judging pass's end): cleared once said. +// SayUnsaid says again each manifest-accepted whose finding stands (the judging pass's end): cleared once said. What stands is +// what the finding writer holds (design 47 G A-38). func (ix *Indexer) SayUnsaid(ctx context.Context) { if ix.Findings == nil { return } - keys, err := ix.Findings.Keys(ctx) - if err != nil { - return - } - for _, key := range keys { - if !strings.HasSuffix(key, "."+UnsaidKind) { - continue - } - value, _, found, err := ix.Findings.Get(ctx, key) - if err != nil || !found { - continue - } - f, err := mcc.DecodeFinding(value) - if err != nil || f.Kind != UnsaidKind { + for _, f := range ix.Findings.Held() { + if f.Kind != UnsaidKind { continue } module := strings.TrimPrefix(f.Subject.ID, Module+".accepted.") diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index cfb56164..d86201b2 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -84,9 +84,9 @@ type Indexer struct { // DLV-I9): for every accepted verdict newly written, carrying the entries whole, and an empty one for a module leaving // the trunk; nil says nothing. SayAccepted func(ctx context.Context, a meshcatalog.ManifestAcceptedEvent, id string) error - // Findings is the indexer's findings state, where a manifest-accepted whose publish failed twice is kept until said - // (accepted.go, DLV-I9b); nil writes none. - Findings Store + // Findings is the indexer's finding writer (design 47 G A-29), where a manifest-accepted whose publish failed twice is kept + // until said (accepted.go, DLV-I9b); nil writes none. + Findings FindingWriter // Packages is the packages state (PackagesBucket, packages.go; novox/hq ADR 0593 §2); nil publishes none, // and every fingerprint takes the version line. Packages Store 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 337fb881..5cb909fa 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records_test.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records_test.go @@ -165,7 +165,7 @@ func TestTheControlRecords(t *testing.T) { m3: {"seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0112", 1)}, head: {"seats/git/seat.json": strings.Replace(git, "ADR 0111", "ADR 0200", 1)}, } - defs, findings := newMem(), newMem() + defs, findings := newMem(), newFindings() var said []SeatDefinitionChanged ix := &Indexer{SeatDefinitions: defs, Findings: findings, Logf: t.Logf, // The forge's listing of a contracts module's `seats/` directory: every blob under it at the commit (ADR 0635 §6). diff --git a/modules/mesh-delivery-indexer/internal/index/findings.go b/modules/mesh-delivery-indexer/internal/index/findings.go new file mode 100644 index 00000000..8e83eae3 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/findings.go @@ -0,0 +1,49 @@ +package index + +import ( + "context" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" +) + +// FindingsState is the indexer's state of its findings, and FindingsBucket its bucket (design 47 G §383, §687). +const ( + FindingsState = "findings" + FindingsBucket = Module + "_" + FindingsState +) + +// FindingWriter is the indexer's finding writer, mesh-sdk's held writer of the seat mesh-conditions' generated finding (design +// 47 G A-29, A-38): Write raises a finding, Clear clears one, Held answers the findings it holds, so the indexer reads what +// stands from the writer. +type FindingWriter interface { + Write(ctx context.Context, f mcseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []mcseat.Finding +} + +// keyID is an id as the parts of a key: each dot-separated part made safe for a bus key as a condition's key is (design 47 G +// A-22), so a finding's key keeps the shape it had. +func keyID(id string) string { + key := mcc.Key("x", id, "x") + return key[len("x.") : len(key)-len(".x")] +} + +// observed is the indexer's finding of kind for the subject id as the seat's generated finding: a warning, its id made safe as a +// key's part, the writer completing its `since` (design 47 G §686, A-22). +func (ix *Indexer) observed(scope, id, kind, summary, headline, explanation, source string) mcseat.Finding { + id = keyID(id) + return mcseat.Finding{Scope: scope, ID: id, Kind: kind, Severity: mcseat.FindingSeverityWarning, Summary: summary, Headline: headline, + Explanation: explanation, Source: source, Observer: Module, At: ix.now().UTC(), Subject: mcseat.Subject{Scope: scope, ID: id}} +} + +// clearFinding clears the finding of scope, id and kind when the writer holds it: a clear of none held says nothing. +func (ix *Indexer) clearFinding(ctx context.Context, scope, id, kind, why string) error { + id = keyID(id) + for _, f := range ix.Findings.Held() { + if f.Scope == scope && f.ID == id && f.Kind == kind { + return ix.Findings.Clear(ctx, scope, id, kind, why) + } + } + return nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/helpers_test.go b/modules/mesh-delivery-indexer/internal/index/helpers_test.go index 4c458ce4..d49cff8f 100644 --- a/modules/mesh-delivery-indexer/internal/index/helpers_test.go +++ b/modules/mesh-delivery-indexer/internal/index/helpers_test.go @@ -6,6 +6,10 @@ import ( "maps" "slices" "sync" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/views" ) // rawList is the list a kept predefined value holds, as it was written. @@ -82,3 +86,42 @@ var errObjectsFullForTest = &objectsFullError{} type objectsFullError struct{} func (*objectsFullError) Error() string { return "maximum bytes exceeded" } + +// memFindings is the finding writer in memory (FindingWriter): what is written is kept by its key as the finding's JSON, what +// is cleared is gone, and Held answers what is kept. +type memFindings struct { + vals map[string][]byte + err error // every write's error +} + +func newFindings() *memFindings { return &memFindings{vals: map[string][]byte{}} } + +func (m *memFindings) Write(_ context.Context, f mcseat.Finding) error { + if m.err != nil { + return m.err + } + raw, err := json.Marshal(f) + m.vals[views.FindingKey(f.Scope, f.ID, f.Kind)] = raw + return err +} + +func (m *memFindings) Clear(_ context.Context, scope, id, kind, _ string) error { + delete(m.vals, views.FindingKey(scope, id, kind)) + return nil +} + +func (m *memFindings) Held() []mcseat.Finding { + var out []mcseat.Finding + for _, k := range slices.Sorted(maps.Keys(m.vals)) { + var f mcseat.Finding + if json.Unmarshal(m.vals[k], &f) == nil { + out = append(out, f) + } + } + return out +} + +// unsaidKey is the key of module's unsaid finding. +func unsaidKey(module string) string { + return views.FindingKey(mcc.ScopeModule, keyID(unsaidID(module)), UnsaidKind) +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index b5d7747c..fc7481fe 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -638,7 +638,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) } { b = held() clock := time.Date(2026, 11, 10, 1, 0, 0, 0, time.UTC) - b.ix.Now, b.ix.Findings = func() time.Time { return clock }, newMem() + b.ix.Now, b.ix.Findings = func() time.Time { return clock }, newFindings() answers := b.ix.Message b.ix.held = nil b.ix.Message = func(context.Context, string, string, int) ([]byte, error) { return nil, c.fault } @@ -653,10 +653,8 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) // machine away, only a toolchain report heard), none either; the next message heard on the forge, the read still // failing, writes call-unanswered naming the bridge's call; the merge read and taken in clears it. finding := func() (string, bool) { - keys, _ := b.ix.Findings.Keys(ctx) - for _, k := range keys { + for k, v := range b.ix.Findings.(*memFindings).vals { if strings.HasSuffix(k, "."+mcc.KindCallUnanswered) { - v, _, _, _ := b.ix.Findings.Get(ctx, k) return string(v), true } } diff --git a/modules/mesh-delivery-indexer/internal/index/seatdefs.go b/modules/mesh-delivery-indexer/internal/index/seatdefs.go index 488bb37f..8d3906c9 100644 --- a/modules/mesh-delivery-indexer/internal/index/seatdefs.go +++ b/modules/mesh-delivery-indexer/internal/index/seatdefs.go @@ -28,6 +28,7 @@ import ( "time" mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" @@ -462,8 +463,7 @@ func (ix *Indexer) twiceFinding(ctx context.Context, repository, module string, fmt.Sprintf("%s's trunk defines seats another trunk already defines, refused: %s", module, strings.Join(lines, "; ")), "A seat on "+module+"'s trunk is already defined by another repository", "A seat has one definition: the one that stood first stays the seat's contract and "+module+"'s is kept nowhere. "+ - "Remove the seat from one of the two repositories; the next merge of "+module+" clears this.", - "No seat on "+module+"'s trunk is defined by another repository.") + "Remove the seat from one of the two repositories; the next merge of "+module+" clears this.") } // outsideFinding writes the contracts module's seat-outside-its-domain finding for the seats it defines that belong to @@ -478,8 +478,7 @@ func (ix *Indexer) outsideFinding(ctx context.Context, module string, outside ma fmt.Sprintf("%s's trunk defines seats of another domain, refused: %s", module, strings.Join(lines, "; ")), "A seat on "+module+"'s trunk belongs to another domain's contracts module", "A seat of the mesh's own contracts is taken only from its domain's contracts module (mesh-contracts for a domain with none). "+ - "Move the definition there; the next merge of "+module+" without it clears this.", - "Every seat on "+module+"'s trunk is of its domain.") + "Move the definition there; the next merge of "+module+" without it clears this.") } // dropPull removes a pull request's entries and its out-of-step files. @@ -528,32 +527,24 @@ func (ix *Indexer) unparsedFinding(ctx context.Context, module string, unparsed fmt.Sprintf("%s's trunk holds seat definitions that do not parse: %s", module, strings.Join(lines, "; ")), "A seat definition on "+module+"'s trunk does not parse", "The trunk's last definition that parsed stays the seat's contract. A change fixing the definition clears this when it "+ - "merges; indexer_seat_definitions_again reads the trunk again now.", - "Every seat definition on "+module+"'s trunk parses.") + "merges; indexer_seat_definitions_again reads the trunk again now.") } -// finding writes (raise) or deletes the indexer's finding of a kind for a subject id. -func (ix *Indexer) finding(ctx context.Context, id, kind string, raise bool, summary, headline, explanation, resolved string) { +// finding raises (raise) or clears the indexer's finding of a kind for a subject id through its finding writer (design 47 G +// §689, A-29): a finding of the seat mesh-conditions' generated type, its id made safe as a key's part (A-22); the writer says +// again only what changed. +func (ix *Indexer) finding(ctx context.Context, id, kind string, raise bool, summary, headline, explanation string) { if ix.Findings == nil { return } - key := mcc.Key(mcc.ScopeModule, id, kind) - err := write(ctx, ix.Findings, key, func(value []byte, found bool) ([]byte, bool, error) { - if !raise { - return nil, found, nil - } - if found { - if f, err := mcc.DecodeFinding(value); err == nil && f.Summary == summary { - return nil, false, nil - } - } - at := ix.now().UTC() - data, err := mcc.EncodeFinding(mcc.Observation{Scope: mcc.ScopeModule, ID: id, Kind: kind, Summary: summary, - Headline: headline, Explanation: explanation, Resolved: resolved, Source: kind}.Finding(Module, "", at)) - return data, false, err - }) + var err error + if raise { + err = ix.Findings.Write(ctx, ix.observed(mcc.ScopeModule, id, kind, summary, headline, explanation, kind)) + } else { + err = ix.clearFinding(ctx, mcc.ScopeModule, id, kind, "the indexer no longer observes it") + } if err != nil { - ix.logf("seat definitions: the finding %s is not written: %v", key, err) + ix.logf("seat definitions: the finding %s is not written: %v", views.FindingKey(mcc.ScopeModule, keyID(id), kind), err) } } @@ -584,10 +575,9 @@ func (ix *Indexer) ApplyOutOfStepBuild(ctx context.Context, value []byte) { fmt.Sprintf("%s's trunk build at %.12s has generated files out of step with their seat definitions: %s", e.Module, e.Commit, strings.Join(files, ", ")), "Generated types on "+e.Module+"'s trunk are out of step with their seat definitions", "Its build's test step regenerated the types and found the committed files differ, missing or not written by the "+ - "generator (a generator's change does this). A change regenerating them clears this once a build of "+e.Module+" ends clean.", - "A build of "+e.Module+" ended clean: its generated types match their definitions.") + "generator (a generator's change does this). A change regenerating them clears this once a build of "+e.Module+" ends clean.") case e.Outcome == builds.BuildDone: - ix.finding(ctx, findingID("out-of-step", e.Module), OutOfStepKind, false, "", "", "", "") + ix.finding(ctx, findingID("out-of-step", e.Module), OutOfStepKind, false, "", "", "") } } @@ -699,8 +689,8 @@ func OutOfStepOf(defs, findings views.Source) ([]OutOfStepRow, error) { if err != nil || !ok || v == nil { continue } - f, err := mcc.DecodeFinding(v.Value) - if err != nil || f.Kind != OutOfStepKind { + var f mcseat.Finding + if err := json.Unmarshal(v.Value, &f); err != nil || f.Kind != OutOfStepKind { continue } out = append(out, OutOfStepRow{Source: SourceTrunk, Module: strings.TrimPrefix(f.Subject.ID, Module+".out-of-step."), Summary: f.Summary}) diff --git a/modules/mesh-delivery-indexer/internal/index/unreadcall.go b/modules/mesh-delivery-indexer/internal/index/unreadcall.go index e8d2f1da..d81e49fb 100644 --- a/modules/mesh-delivery-indexer/internal/index/unreadcall.go +++ b/modules/mesh-delivery-indexer/internal/index/unreadcall.go @@ -2,9 +2,13 @@ package index import ( "context" + "fmt" "strings" + "time" mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/views" ) // **A merged message unread past five minutes, the forge heard, is a finding** (issue 1966 note 153; novox/hq ADR 0668 §3). @@ -21,7 +25,8 @@ const UnreadCallee = ForgeBridgeModule + ".bridge_merged_message" // unreadCallID is the finding's id for the pending key: the merge it waits on. func unreadCallID(key string) string { return "merged-message." + key } -// unreadFindings raises the finding of every merge of entries unread past the bound and clears the finding of every other. +// unreadFindings raises the finding of every merge of entries unread past the bound and clears the finding of every other +// that the finding writer holds (design 47 G §689, A-29). func (ix *Indexer) unreadFindings(ctx context.Context, entries []PendingEntry) { if ix.Findings == nil { return @@ -31,39 +36,32 @@ func (ix *Indexer) unreadFindings(ctx context.Context, entries []PendingEntry) { if e.Value.Unread == "" { continue } - unread[mcc.Key(mcc.ScopeCall, unreadCallID(e.Key), mcc.KindCallUnanswered)] = true + id := keyID(unreadCallID(e.Key)) + unread[id] = true since := e.Value.UnreadSince if since.IsZero() || ix.now().Sub(since) < mcc.CallUnansweredAfter { continue } - o := mcc.CallUnanswered(unreadCallID(e.Key), UnreadCallee, "", e.Value.Unread, "indexer."+mcc.KindCallUnanswered, since) - o.Needs = TakeInPendingTool + " reads the merge again now; " + PendingTool + " shows it unread." - key := o.Key() - if err := write(ctx, ix.Findings, key, func(value []byte, found bool) ([]byte, bool, error) { - if found { - if f, err := mcc.DecodeFinding(value); err == nil && f.Summary == o.Summary { - return nil, false, nil - } - } - data, err := mcc.EncodeFinding(o.Finding(Module, "", ix.now().UTC())) - return data, false, err - }); err != nil { - ix.logf("take-in: the finding %s is not written: %v", key, err) + f := ix.observed(mcc.ScopeCall, unreadCallID(e.Key), mcc.KindCallUnanswered, + fmt.Sprintf("the call %s about %s is unanswered since %s", UnreadCallee, unreadCallID(e.Key), since.UTC().Format(time.RFC3339)), + "A call whose answer is kept is not answered", + fmt.Sprintf("The call %s is made again on every transport fault and on the callee's event; it has not been answered since %s. Its last fault: %s.", + UnreadCallee, since.UTC().Format(time.RFC3339), e.Value.Unread), "indexer."+mcc.KindCallUnanswered) + f.Needs = TakeInPendingTool + " reads the merge again now; " + PendingTool + " shows it unread." + f.Evidence = []mcseat.Evidence{{At: since.UTC(), Said: e.Value.Unread}} + if err := ix.Findings.Write(ctx, f); err != nil { + ix.logf("take-in: the finding %s is not written: %v", views.FindingKey(f.Scope, f.ID, f.Kind), err) } else { - ix.logf("take-in: %s (%s#%d) is unread since %s and the forge is heard: the finding %s", e.Key, e.Forge.Repository, e.Forge.Number, since.Format("2006-01-02T15:04:05Z07:00"), key) + ix.logf("take-in: %s (%s#%d) is unread since %s and the forge is heard: the finding %s", e.Key, e.Forge.Repository, e.Forge.Number, since.Format("2006-01-02T15:04:05Z07:00"), views.FindingKey(f.Scope, f.ID, f.Kind)) } } - keys, err := ix.Findings.Keys(ctx) - if err != nil { - return - } - prefix, suffix := mcc.ScopeCall+"."+unreadCallID(""), "."+mcc.KindCallUnanswered - for _, k := range keys { - if !strings.HasPrefix(k, prefix) || !strings.HasSuffix(k, suffix) || unread[k] { + prefix := keyID(strings.TrimSuffix(unreadCallID(""), ".")) + "." + for _, f := range ix.Findings.Held() { + if f.Scope != mcc.ScopeCall || f.Kind != mcc.KindCallUnanswered || !strings.HasPrefix(f.ID, prefix) || unread[f.ID] { continue } - if err := write(ctx, ix.Findings, k, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }); err != nil { - ix.logf("take-in: the finding %s is not cleared: %v", k, err) + if err := ix.Findings.Clear(ctx, f.Scope, f.ID, f.Kind, "the merge is no longer unread"); err != nil { + ix.logf("take-in: the finding %s is not cleared: %v", views.FindingKey(f.Scope, f.ID, f.Kind), err) } } } diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go index 87353e68..b72ad051 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go @@ -237,7 +237,7 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) { } // DLV-I9b: a failed publish is published again; one failing twice writes the finding naming the module, which the next // judging pass says again and clears. - ix.Findings = newMem() + ix.Findings = newFindings() failing = 1 once := hold("once", `{"module":"once","version":"1.0"}`) if _, err := ix.JudgeAll(ctx, map[string]bool{once: true}); err != nil || accepted[len(accepted)-1].Module != "once" { @@ -248,13 +248,13 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) { if _, err := ix.JudgeAll(ctx, map[string]bool{flaky: true}); err != nil { t.Fatal(err) } - if value, _, found, _ := ix.Findings.Get(ctx, unsaidKey("flaky")); !found || !strings.Contains(string(value), "flaky") { + if value, found := ix.Findings.(*memFindings).vals[unsaidKey("flaky")]; !found || !strings.Contains(string(value), "flaky") { t.Fatalf("a publish failing twice wrote no finding naming flaky: %s", value) } if _, err := ix.JudgeAll(ctx, nil); err != nil { t.Fatal(err) } - if _, _, found, _ := ix.Findings.Get(ctx, unsaidKey("flaky")); found || accepted[len(accepted)-1].Module != "flaky" || + if _, found := ix.Findings.(*memFindings).vals[unsaidKey("flaky")]; found || accepted[len(accepted)-1].Module != "flaky" || accepted[len(accepted)-1].Digest != flaky { t.Fatalf("the next pass did not say flaky and clear its finding (found %v, last said %+v)", found, accepted[len(accepted)-1]) } diff --git a/modules/mesh-delivery-indexer/internal/index/views.go b/modules/mesh-delivery-indexer/internal/index/views.go index c57952ff..9aacd256 100644 --- a/modules/mesh-delivery-indexer/internal/index/views.go +++ b/modules/mesh-delivery-indexer/internal/index/views.go @@ -1,8 +1,6 @@ package index import ( - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" - "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -30,7 +28,7 @@ var ReadTools = []ViewedTool{ {Tool: "indexer_read_graph", State: graph.State, Words: "the module graph: each module's build and provision edges."}, {Tool: "indexer_read_heard", State: HeardState, Words: "the forge messages the indexer heard and what it did with each."}, {Tool: "indexer_read_taken", State: "taken", Words: "each repository's merge the indexer took in last."}, - {Tool: "indexer_read_findings", State: mcc.FindingsState, Words: "the indexer's findings, which the condition keeper raises and clears."}, + {Tool: "indexer_read_findings", State: FindingsState, Words: "the indexer's findings, which the condition keeper raises and clears."}, {Tool: "indexer_read_seat_definitions", State: SeatDefinitionsState, Words: "the seat definitions the trunk's and each open pull request's contracts modules hold, resolved, or why one does not parse."}, {Tool: "indexer_read_unjudged", State: UnjudgedState, Words: "the manifests a take-in newly wrote that no judge pass judged yet, by digest, each with its module and merge; indexer_judge judges them."}, } @@ -50,7 +48,7 @@ var ViewedStates = map[string]string{ PackagesState: PackagesBucket, HeardState: HeardBucket, "taken": TakenBucket, - mcc.FindingsState: mcc.FindingsBucket(Module), + FindingsState: FindingsBucket, SeatDefinitionsState: SeatDefinitionsBucket, UnjudgedState: UnjudgedBucket, } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index fb11fcfd..f22b616d 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -16,7 +16,6 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "git.novox.be/novox/mesh-sdk/go/catalogue" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-indexer/internal/index" @@ -121,7 +120,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: index.PackagesState, History: index.PackagesHistory}, - {Name: views.MarksState, History: 1}, {Name: mcc.FindingsState, History: 5}, {Name: views.StatusState, History: 1}, + {Name: views.MarksState, History: 1}, {Name: index.FindingsState, History: 5}, {Name: views.StatusState, History: 1}, {Name: index.RequirementsState, History: 5}, {Name: index.SeatDefinitionsState, History: 5}, {Name: index.PendingState, History: 1}, {Name: index.UnjudgedState, History: 1}, {Name: index.EntryIndexState, History: 1}, {Name: index.RegistrationsState, History: 1}, {Name: index.ProposalsState, History: 1}}; !reflect.DeepEqual(m.State, want) { // DLV-I7, DLV-I17