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 e40a2169..9e558938 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -451,7 +451,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, // Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it // is not there yet, a node is asked again only when it comes back online. if js != nil { - hearReadsChanged(ctx, js, reads, logger.Printf) + hearReadsChanged(ctx, js, reads, findings, logger.Printf) } if bus != nil && kvs[delivery.AssignerPutsBucket] != nil { p.acts = newPutActs(bus, wake, logger.Printf) @@ -520,12 +520,13 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro // hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's durable consumer on EVENTS, // and asks the node it names again. Any other event the consumer hands is acknowledged and left. -func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *enginereads.Keeper, logf func(string, ...any)) { +func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) { account := process.Account{Node: thisNode, Module: delivery.AssignerModule} + w := notHeardWriter(findings, logf) cons, err := account.EventConsumer(ctx, js, process.EventsStream) if err != nil { - logf("the consumer %s on %s does not exist yet, so %s is not heard until the assigner starts again (a node is asked again when it comes back online): %v", - account.EventConsumerName(), process.EventsStream, enginereads.Event, err) + w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again (a node is asked again when it comes back online)", + fmt.Errorf("the consumer %s on %s does not exist yet: %w", account.EventConsumerName(), process.EventsStream, err)) return } cc, err := process.HearEvents(cons, func(m jetstream.Msg) { @@ -543,9 +544,10 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, reads *engine } }, logf) if err != nil { - logf("%s is not heard: %v", enginereads.Event, err) + w.NotHeard(process.EventsStream, enginereads.Event+" is not heard until the assigner starts again", err) return } + w.Heard(process.EventsStream) go func() { <-ctx.Done() cc.Stop() diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 14197f0c..cba4f363 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -3,6 +3,8 @@ package main import ( "context" "encoding/json" + "fmt" + "slices" "time" "github.com/nats-io/nats.go/jetstream" @@ -17,6 +19,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/views" "git.novox.be/novox/mesh-delivery/sdk/graph" + "git.novox.be/novox/mesh-delivery/sdk/heard" "mesh-delivery-assigner/internal/assign" ) @@ -186,12 +189,30 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle } }) } + // Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5). stream := catalogue.StreamBusName(indexerViewsStream) - if cons, err := iv.account.EventConsumer(ctx, iv.js, stream); err != nil { - iv.logf("the consumer %s on %s does not exist yet (%v): the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", - iv.account.EventConsumerName(), stream, err) - } else if _, err := process.HearEvents(cons, views.Hear(ctx, followers, iv.logf), iv.logf); err != nil { - iv.logf("the indexer's views family is not heard (%v) until the assigner starts again", err) - } + _ = notHeardWriter(iv.findings, iv.logf).Bind(stream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error { + cons, err := iv.account.EventConsumer(ctx, iv.js, stream) + if err != nil { + return fmt.Errorf("the consumer %s on %s does not exist yet: %w", iv.account.EventConsumerName(), stream, err) + } + _, err = process.HearEvents(cons, views.Hear(ctx, followers, iv.logf), iv.logf) + return err + }) return len(all) } + +// notHeardWriter writes the assigner's finding of a family it does not hear into its findings (heard.Writer); with no +// findings state it is said in the log alone. +func notHeardWriter(findings assign.Findings, logf func(string, ...any)) heard.Writer { + w := heard.Writer{Observer: delivery.AssignerModule, Logf: logf, Put: findings.Put} + if findings.Delete != nil { + w.Delete = func(key string) error { + if keys, err := findings.Keys(); err == nil && !slices.Contains(keys, key) { + return nil // none stands + } + return findings.Delete(key) + } + } + return w +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go index 7fe0f73a..6c458d2c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go @@ -11,6 +11,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-delivery/sdk/heard" + "mesh-delivery-builder/internal/versions" ) @@ -122,6 +124,39 @@ func (f *findings) clear(ctx context.Context, kind, scope, id string) { } } +// notHeard is the writer of the builder's finding of a family it does not hear (heard.Writer, ADR 0615 §5): written into +// its findings once bound, said in the log until then. +func (f *findings) notHeard(ctx context.Context) heard.Writer { + on := func(do func(findingStore)) { + switch { + case f.store != nil: + do(f.store) + case f.kv != nil: + f.kv.Ready(func(kv jetstream.KeyValue) { do(kv) }) + } + } + return heard.Writer{Observer: delivery.BuilderModule, Logf: f.logf, Now: f.now, + Put: func(key string, value []byte) error { + on(func(kv findingStore) { + if _, err := kv.Put(ctx, key, value); err != nil { + f.logf("the finding %s cannot be written: %v", key, err) + } + }) + return nil + }, + Delete: func(key string) error { + on(func(kv findingStore) { + if _, err := kv.Get(ctx, key); err != nil { + return // none stands + } + if err := kv.Delete(ctx, key); err != nil { + f.logf("the finding %s cannot be cleared: %v", key, err) + } + }) + return nil + }} +} + // floorSaid is the floor finding (design 51 *Health and conditions*: a floor source silent): raised for a name waiting on // its floor, about that name, cleared once all three sources answered. func (f *findings) floorSaid(ctx context.Context) func(versions.FloorWait, bool) { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 54096a70..d58246d3 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -5,6 +5,7 @@ package main import ( + "cmp" "context" "fmt" "io" @@ -292,42 +293,44 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A build module says each time it wrote its toolchains key: its stream, and the builder's durable consumer on it, are // made by the controller. Where the stream is not there yet the builder says so and is started again with the build // module. + // Each consumer that is not there yet is the finding family-not-heard naming its stream until it binds at a later + // start (heard.Writer; ADR 0615 §5); one there that cannot be heard stops the builder, as before. + notHeard := found.notHeard(ctx) + var stopped error + hear := func(stream, consequence string, each func(jetstream.Msg)) { + _ = notHeard.Bind(stream, consequence, func() error { + cons, err := cred.EventConsumer(ctx, js, stream) + if err != nil { + return fmt.Errorf("the consumer %s_%s on %s does not exist yet: %w", cred.Node, cred.Module, stream, err) + } + cc, err := process.HearEvents(cons, each, logger.Printf) + if err != nil { + stopped = cmp.Or(stopped, err) + return err + } + go func() { <-ctx.Done(); cc.Stop() }() + return nil + }) + } for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} { - stream := toolchainStreams[module] - cons, err := cred.EventConsumer(ctx, js, stream) - if err != nil { - logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the builder starts again: %v", cred.Node, cred.Module, stream, module, err) - continue - } - cc, err := process.HearEvents(cons, func(m jetstream.Msg) { watch.Reported(m) }, logger.Printf) - if err != nil { - return err - } - defer cc.Stop() + hear(toolchainStreams[module], module+"'s toolchain reports are not heard until the builder starts again", func(m jetstream.Msg) { watch.Reported(m) }) } // The seat mesh-artifact-sweep says artifacts-gone when it found a key's blob gone (issue 1937, ADR 0571 §2): the // builder's durable consumer on EVENTS, made by the grants part from the manifest's consumes. Where it is not there yet the builder says // so; it still asks held at its start and at every module value. - switch cons, err := cred.EventConsumer(ctx, js, eventsStream); { - case err != nil: - logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v", - cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err) - default: - // The same consumer carries what wakes a floor (design 51 *The floor*): the indexer's manifest-accepted and the - // condition keeper's condition-cleared. - floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf} - // And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version. - seats := seatHearer{n: numberer, b: b, logf: logger.Printf} - cc, err := process.HearEvents(cons, func(m jetstream.Msg) { - if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4) - return - } - heardGone(ctx, b, m, logger.Printf) - }, logger.Printf) - if err != nil { - return err + // The same consumer carries what wakes a floor (design 51 *The floor*): the indexer's manifest-accepted and the + // condition keeper's condition-cleared. + floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf} + // And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version. + seats := seatHearer{n: numberer, b: b, logf: logger.Printf} + hear(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words and manifest-reads-changed are not heard until the builder starts again", func(m jetstream.Msg) { + if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4) + return } - defer cc.Stop() + heardGone(ctx, b, m, logger.Printf) + }) + if stopped != nil { + return stopped } reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "builder_first_views reads nothing and builder_first_view_again is not heard", logger.Printf)) // The modules, once their first views and the manifests' are answered: what a module builds is read from the manifests. @@ -396,16 +399,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro tagWanted(ctx, tagAsksKV.KV(), m, delivery.CheckerModule, time.Now(), logger.Printf) }}, } { - cons, err := cred.EventConsumer(ctx, js, c.stream) - if err != nil { - logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v", cred.Node, cred.Module, c.stream, c.what, err) - continue - } - cc, err := process.HearEvents(cons, c.hear, logger.Printf) - if err != nil { - return err - } - defer cc.Stop() + hear(c.stream, c.what+" is not heard until the builder starts again", c.hear) + } + if stopped != nil { + return stopped } // The state engine-reads, new with ADR 0618 (issue 1790's kind): until it is on the bus each statement is kept in memory // only, and builder_engine_reads answers nothing. diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index e98037ab..a7305b5c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -13,6 +13,8 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-delivery/sdk/heard" "git.novox.be/novox/mesh-sdk/go/process" "github.com/nats-io/nats.go" @@ -338,6 +340,17 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { if _, ok := st.vals[floorKey]; ok { t.Error("a known floor leaves floor-silent") } + // ADR 0615 §5 (review round 1): a consumer not there yet is family-not-heard naming its stream, cleared when it binds. + notHeard := f.notHeard(context.Background()) + unheardKey := heard.Key(delivery.BuilderModule, "FORGE") + _ = notHeard.Bind("FORGE", "a branch's end is not heard", func() error { return errors.New("consumer not found") }) + if got, err := health.DecodeFinding(st.vals[unheardKey]); err != nil || got.Kind != heard.Kind || !strings.Contains(got.Summary, "does not hear FORGE") { + t.Fatalf("a consumer not there: the finding = %+v, %v", got, err) + } + _ = notHeard.Bind("FORGE", "a branch's end is not heard", func() error { return nil }) + if _, ok := st.vals[unheardKey]; ok { + t.Error("a consumer bound leaves family-not-heard") + } // DLV-B9, the old form's window: one counted node reading only the first form holds it; seven days on it is // old-form-held about that node; the node stating the newest form closes it, said once, the finding cleared; a // retired node dropped by hand leaves the count. 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 8c4ab17b..b04f2850 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -9,6 +9,7 @@ package main import ( + "cmp" "context" "encoding/json" "errors" @@ -468,58 +469,44 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // 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 // 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) + // 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)) { + _ = notHeard.Bind(stream, consequence, func() error { + cons, err := cred.EventConsumer(ctx, js, stream) + if err != nil { + return fmt.Errorf("the consumer %s_%s on %s does not exist yet: %w", cred.Node, cred.Module, stream, err) + } + hc, err := process.HearEvents(cons, each, logger.Printf) + if err != nil { + stopped = cmp.Or(stopped, err) + return err + } + go func() { <-ctx.Done(); hc.Stop() }() + return nil + }) + } for _, module := range sortedModules() { - stream := index.ToolchainStreams[module] - tcons, err := cred.EventConsumer(ctx, js, stream) - if err != nil { - logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the indexer starts again: %v", cred.Node, cred.Module, stream, module, err) - continue - } - tc, err := process.HearEvents(tcons, func(m jetstream.Msg) { tools.Reported(m) }, logger.Printf) - if err != nil { - return err - } - defer tc.Stop() - } - switch pcons, err := cred.EventConsumer(ctx, js, PreviewsStream); { - case err != nil: - logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's previews are not heard until the indexer starts again: %v", cred.Node, cred.Module, PreviewsStream, err) - default: - pc, err := process.HearEvents(pcons, func(m jetstream.Msg) { previews.Written(m) }, logger.Printf) - if err != nil { - return err - } - defer pc.Stop() + hear(index.ToolchainStreams[module], module+"'s toolchain reports are not heard until the indexer starts again", func(m jetstream.Msg) { tools.Reported(m) }) } + hear(PreviewsStream, "the checker's previews are not heard until the indexer starts again", func(m jetstream.Msg) { previews.Written(m) }) // The versions each package's series holds (DLV-I5): the registry seats' published and the builder's numbered, on the // indexer's durable consumer of the shared events stream. Where it is not there yet, no series moves until the indexer // starts again, said. - switch vcons, err := cred.EventConsumer(ctx, js, process.EventsStream); { - case err != nil: - logger.Printf("the consumer %s_%s on %s does not exist yet, so no package's series moves until the indexer starts again: %v", cred.Node, cred.Module, process.EventsStream, err) - default: - vc, err := process.HearEvents(vcons, func(m jetstream.Msg) { - if !heardVersion(ctx, ix, m, logger.Printf) { - _ = m.Ack() // nothing else the indexer consumes rides the shared stream - } - }, logger.Printf) - if err != nil { - return err + hear(process.EventsStream, "no package's series moves until the indexer starts again", func(m jetstream.Msg) { + if !heardVersion(ctx, ix, m, logger.Printf) { + _ = m.Ack() // nothing else the indexer consumes rides the shared stream } - defer vc.Stop() - } + }) // The checker's verdicts carry a pull request's out-of-step files (DLV-I20): on the indexer's durable consumer of the // checker's stream asks. Where it is not there yet, no pull request's out-of-step files are kept until the indexer // starts again, said. - switch acons, err := cred.EventConsumer(ctx, js, checkerAsksStream); { - case err != nil: - logger.Printf("the consumer %s_%s on %s does not exist yet, so no pull request's out-of-step files are kept until the indexer starts again: %v", cred.Node, cred.Module, checkerAsksStream, err) - default: - ac, err := process.HearEvents(acons, func(m jetstream.Msg) { heardVerdict(ctx, ix, m, logger.Printf) }, logger.Printf) - if err != nil { - return err - } - defer ac.Stop() + hear(checkerAsksStream, "no pull request's out-of-step files are kept until the indexer starts again", func(m jetstream.Msg) { heardVerdict(ctx, ix, m, logger.Printf) }) + if stopped != nil { + return stopped } cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { 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 9d43ea9b..0e9e6d28 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" + "fmt" "time" "github.com/nats-io/nats.go/jetstream" @@ -14,6 +15,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-delivery/sdk/graph" + "git.novox.be/novox/mesh-delivery/sdk/heard" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -72,6 +74,30 @@ 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 { + return heard.Writer{Observer: delivery.IndexerModule, 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 + }, + 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 + }} +} + // graphState is the module graph's state, named apart from run's bucket of the same name. const graphState = graph.State @@ -131,16 +157,17 @@ func followPlacement(ctx context.Context, js jetstream.JetStream, cred process.A } close(answered) }() - // The assigner's views family, on the indexer's durable consumer, once the first view is asked. - switch cons, err := cred.EventConsumer(ctx, js, catalogue.StreamBusName(assignerViewsStream)); { - case err != nil: - logf("the consumer %s on %s does not exist yet (%v): the placement is the first view's alone, asked again by hand, until the indexer starts again", - cred.EventConsumerName(), catalogue.StreamBusName(assignerViewsStream), err) - default: - if _, err := process.HearEvents(cons, views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf), logf); err != nil { - logf("the assigner's views family is not heard (%v): the placement is the first view's alone until the indexer starts again", err) + // The assigner's views family, on the indexer's durable consumer, once the first view is asked; not heard, it is the + // finding family-not-heard until it binds (ADR 0615 §5). + stream := catalogue.StreamBusName(assignerViewsStream) + _ = notHeardWriter(ctx, findings, logf).Bind(stream, "the placement is the first view's alone, asked again by hand, until the indexer starts again", func() error { + cons, err := cred.EventConsumer(ctx, js, stream) + if err != nil { + return fmt.Errorf("the consumer %s on %s does not exist yet: %w", cred.EventConsumerName(), stream, err) } - } + _, err = process.HearEvents(cons, views.Hear(ctx, map[string]*views.Follower{delivery.PlacementState: f}, logf), logf) + return err + }) wait, cancel := context.WithTimeout(ctx, views.Bound) // a context's bound, no timer (ADR 0437 §8) defer cancel() select { diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index 12cfabb3..52bd98d8 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -67,31 +67,38 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco if opened[stream] { return } + // Not bound, the stream is the finding family-not-heard (ADR 0615 §5) until it is, at a later change of the + // membership or a later start. cons, err := js.Consumer(ctx, stream, name) if err != nil { - logf("the consumer %s on %s cannot be bound yet (%v): what it carries is heard once the grants part has made it", name, stream, err) + w.NotHeard(stream, "what it carries is heard once the grants part has made the consumer "+name, err) return } if _, err := process.HearEvents(cons, func(m jetstream.Msg) { consumed(w, m.Subject(), m.Data(), logf) _ = m.Ack() }, logf); err != nil { - logf("the consumer %s on %s cannot be heard (%v)", name, stream, err) + w.NotHeard(stream, "the consumer "+name+" cannot be heard", err) return } + w.Heard(stream) opened[stream] = true } open("EVENTS", cred.EventConsumerName()) + // Its own membership, whose change opens a consumer not there yet: not read or not watched, it is the finding + // family-not-heard naming it, until the watchman starts again and watches it. + const membershipFamily = "membership" kv, err := js.KeyValue(ctx, membership.GrantsBucket) if err != nil { - logf("its own membership cannot be read (%v): the bus module's created and refused are not heard until the watchman starts again", err) + w.NotHeard(membershipFamily, "a consumer not there yet is not opened, so the bus module's created and refused may go unheard until the watchman starts again", err) return } watcher, err := kv.Watch(ctx, membership.GrantsKey(cred.Node, cred.Module)) if err != nil { - logf("its own membership cannot be watched (%v): the bus module's created and refused are not heard until the watchman starts again", err) + w.NotHeard(membershipFamily, "a consumer not there yet is not opened, so the bus module's created and refused may go unheard until the watchman starts again", err) return } + w.Heard(membershipFamily) go func() { defer watcher.Stop() for e := range watcher.Updates() { diff --git a/modules/mesh-delivery-watchman/internal/watch/finding.go b/modules/mesh-delivery-watchman/internal/watch/finding.go index d3d435c0..abf0af0c 100644 --- a/modules/mesh-delivery-watchman/internal/watch/finding.go +++ b/modules/mesh-delivery-watchman/internal/watch/finding.go @@ -4,6 +4,8 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-delivery/sdk/heard" ) // The watchman is an observer of the seat mesh-watchdogs (novox/hq ADR 0564 §2, §4): what it observes wrong it writes as a @@ -157,3 +159,22 @@ func (w *Watchman) Rewrite() int { } return n } + +// NotHeard is the watchman's finding of a family it does not hear (heard.Kind, ADR 0615 §5): a consumer not there yet, a +// hearing or a watch that could not start; consequence says what goes unheard. Heard ends it, once the consumer binds. +func (w *Watchman) NotHeard(family, consequence string, why error) { + w.mu.Lock() + defer w.mu.Unlock() + summary, headline, explanation := heard.Words(Observer, family, consequence, why) + w.say("%s", summary) + w.find(Finding{Scope: ScopeModule, ID: heard.ID(Observer, family), Token: heard.Kind, Kind: heard.Kind, Severity: Warning, + Source: Observer + "." + heard.Kind, Since: w.now().UTC().Truncate(time.Second), Summary: summary, Headline: headline, + Explanation: explanation, Resolved: "the watchman's consumer of " + family + " is bound"}) +} + +// Heard ends the finding that the watchman does not hear family. +func (w *Watchman) Heard(family string) { + w.mu.Lock() + defer w.mu.Unlock() + w.unfind(heard.Key(Observer, family)) +} diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 097b249a..a3d1c906 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -1,6 +1,7 @@ package watch import ( + "errors" "strings" "sync/atomic" "testing" @@ -8,6 +9,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/heard" "git.novox.be/novox/mesh-sdk/go/provisions" ) @@ -140,6 +143,16 @@ func TestTheWatchmansFindings(t *testing.T) { w.Accepted(accepted("gone")) w.BrokerRefused(BrokerSaid{Module: "mesh-knowledge", Local: "gone"}) }, map[string]string{refusedKey("mesh-knowledge", "gone"): Urgent}}, + {"ADR 0615 §5 (review round 1): a consumer the watchman cannot bind is family-not-heard naming its stream, until it binds", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.NotHeard("EVENTS", "what it carries is heard once the grants part has made the consumer n1_mesh-delivery-watchman", errors.New("consumer not found")) + f := o.findings[heard.Key(Observer, "EVENTS")] + if f.Kind != heard.Kind || f.Severity != Warning || !strings.Contains(f.Summary, "does not hear EVENTS") || !strings.Contains(f.Summary, "consumer not found") { + t.Fatalf("findings = %v", o.findings) + } + w.NotHeard("membership", "a consumer not there yet is not opened", errors.New("no bucket")) + w.Heard("EVENTS") + w.Heard("membership") + }, map[string]string{}}, {"ADR 0601 §4: what a node keeps of a module as it runs, for a manifest that does not say it yet, is a warning until the have says none", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { have := func(kept ...string) []byte { b, err := delivery.EncodeHave(delivery.Have{FP: "src2:" + strings.Repeat("a", 64), Phase: delivery.PhaseRunning, Health: delivery.HealthOK, Kept: kept}) diff --git a/sdk/heard/heard.go b/sdk/heard/heard.go new file mode 100644 index 00000000..06adeabe --- /dev/null +++ b/sdk/heard/heard.go @@ -0,0 +1,109 @@ +// Package heard is the finding a part of mesh-delivery writes while one of its consumers or watches does not hear the +// family it follows (novox/hq ADR 0615 §5; review round 1 of the registry milestone): the consumer not there yet, a +// hearing that could not start, a watch that cannot be opened. Without it the part's health stays healthy and its +// status says answered while the events it waits on are lost to it. The finding names the observer and the family, +// says what is not heard and why, and is deleted when the consumer binds (at the part's next start, or when it opens +// later); the condition keeper raises and clears the condition from it (ADR 0564 §2). +package heard + +import ( + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// Kind is the finding's kind. +const Kind = "family-not-heard" + +// ID is the finding's subject: `.`, the family by the name the part follows it under (a stream, an +// event, a state's watch). +func ID(observer, family string) string { return observer + "." + family } + +// Key is the finding's key in the observer's findings state. +func Key(observer, family string) string { return health.FindingKey(health.ScopeModule, ID(observer, family), Kind) } + +// Words are the finding's words: its summary, headline and explanation. consequence says what the observer misses while +// it does not hear family ("the placement is the first view's alone"). +func Words(observer, family, consequence string, why error) (summary, headline, explanation string) { + reason := "it gave no reason" + if why != nil { + reason = strings.TrimSpace(why.Error()) + } + summary = fmt.Sprintf("%s does not hear %s: %s (%s)", observer, family, consequence, reason) + headline = fmt.Sprintf("%s does not hear %s", observer, family) + explanation = fmt.Sprintf("%s follows %s on a consumer of its own, and that consumer is not there or could not be heard, so %s. "+ + "It is heard again once the consumer binds, at %s's next start or when the grants part has made it; this clears then. "+ + "Look at whether the grants part made %s's consumers.", observer, family, consequence, observer, observer) + return summary, headline, explanation +} + +// Finding is observer's finding that it does not hear family, as health.Finding encodes it. +func Finding(observer, family, consequence string, why error, at time.Time) (string, []byte, error) { + key := Key(observer, family) + summary, headline, explanation := Words(observer, family, consequence, why) + value, err := health.EncodeFinding(health.Finding{Key: key, Kind: Kind, Subject: health.Subject{Scope: health.ScopeModule, ID: ID(observer, family)}, + Severity: health.Warning, Summary: summary, Headline: headline, Explanation: explanation, + Evidence: []health.Evidence{{At: at.UTC(), Said: summary}}, Source: observer + "." + Kind, Observer: observer, At: at.UTC()}) + return key, value, err +} + +// Writer writes and deletes the finding in the observer's findings state. Put or Delete nil writes nothing, said in the +// log, so the part still says it. +type Writer struct { + Observer string + Put func(key string, value []byte) error + Delete func(key string) error + Logf func(string, ...any) + Now func() time.Time +} + +// Bind binds the observer's consumer of family (bind makes it and starts hearing it): the finding is deleted when it +// binds, and written, with why, when it does not. It answers bind's error, so a caller that cannot go on without the +// family still stops. +func (w Writer) Bind(family, consequence string, bind func() error) error { + if err := bind(); err != nil { + w.NotHeard(family, consequence, err) + return err + } + w.Heard(family) + return nil +} + +// NotHeard says in the log and in the finding that the observer does not hear family. +func (w Writer) NotHeard(family, consequence string, why error) { + now := time.Now + if w.Now != nil { + now = w.Now + } + summary, _, _ := Words(w.Observer, family, consequence, why) + w.logf("%s", summary) + if w.Put == nil { + w.logf("the finding %s is not written: no findings state", Key(w.Observer, family)) + return + } + key, value, err := Finding(w.Observer, family, consequence, why, now()) + if err == nil { + err = w.Put(key, value) + } + if err != nil { + w.logf("the finding %s is not written: %v", key, err) + } +} + +// Heard deletes the finding: the consumer of family is bound. +func (w Writer) Heard(family string) { + if w.Delete == nil { + return + } + if err := w.Delete(Key(w.Observer, family)); err != nil { + w.logf("the finding %s is not deleted: %v", Key(w.Observer, family), err) + } +} + +func (w Writer) logf(format string, args ...any) { + if w.Logf != nil { + w.Logf(format, args...) + } +} diff --git a/sdk/heard/heard_test.go b/sdk/heard/heard_test.go new file mode 100644 index 00000000..0c5ed6d9 --- /dev/null +++ b/sdk/heard/heard_test.go @@ -0,0 +1,54 @@ +package heard + +import ( + "errors" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// Review round 1 of the registry milestone (ADR 0615 §5): a consumer that does not bind is a finding naming the observer +// and the family, with why; binding deletes it; no findings state still says it in the log. +func TestAConsumerNotHeardIsAFindingUntilItBinds(t *testing.T) { + at := time.Date(2026, 11, 4, 17, 0, 0, 0, time.UTC) + for _, row := range []struct { + name string + bindErr error + noState bool + want string // the finding's summary, or "" when none stands + }{ + {name: "not there yet", bindErr: errors.New("consumer not found"), + want: "mesh-delivery-indexer does not hear assigner-views: the placement is the first view's alone (consumer not found)"}, + {name: "bound"}, + {name: "no findings state", bindErr: errors.New("consumer not found"), noState: true}, + } { + kept := map[string][]byte{Key("mesh-delivery-indexer", "assigner-views"): []byte("{}")} + var logs []string + w := Writer{Observer: "mesh-delivery-indexer", Now: func() time.Time { return at }, + Logf: func(f string, a ...any) { logs = append(logs, f) }} + if !row.noState { + w.Put = func(k string, v []byte) error { kept[k] = v; return nil } + w.Delete = func(k string) error { delete(kept, k); return nil } + } + err := w.Bind("assigner-views", "the placement is the first view's alone", func() error { return row.bindErr }) + if !errors.Is(err, row.bindErr) { + t.Errorf("%s: Bind answered %v, want %v", row.name, err, row.bindErr) + } + raw, stands := kept[Key("mesh-delivery-indexer", "assigner-views")] + switch { + case row.noState: + if len(logs) == 0 { + t.Errorf("%s: nothing said", row.name) + } + case row.want == "" && stands: + t.Errorf("%s: the finding stands once bound", row.name) + case row.want != "": + f, err := health.DecodeFinding(raw) + if err != nil || f.Kind != Kind || f.Summary != row.want || f.Observer != "mesh-delivery-indexer" || !strings.Contains(f.Subject.ID, "assigner-views") { + t.Errorf("%s: the finding %+v (%v), want summary %q", row.name, f, err, row.want) + } + } + } +}