diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index db989e12..584b29b2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -21,11 +21,17 @@ import ( oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) // 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them // (`.` read as the bucket `_`), so it is granted what it reads and reads nothing else. func TestOpensWhatItsManifestNames(t *testing.T) { + // Review round 2: the assigner's views emitter writes its finding view-unsaid (Emitter.Findings set) when it keeps findings. + findings := assign.Findings{Put: func(string, []byte) error { return nil }, Delete: func(string) error { return nil }} + if viewsEmitter(nil, process.Account{}, nil, findings, nil, t.Logf).Findings == nil { + t.Error("the assigner's views emitter writes no view-unsaid finding") + } // 1297/B2a: without a credential the program does not start, and says why. if err := run(context.Background(), func(string) string { return "" }, io.Discard); err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { t.Errorf("run without a credential = %v", err) 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 75a6ae2e..ca7ae4f1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -98,7 +98,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib } } process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) { - em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.StreamSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf} + em := viewsEmitter(js, account, mk, findings, tooLarge, logf) for _, state := range []string{"asks", "differences", health.FindingsState, delivery.PlacementState, delivery.SeatHoldersState, "wants"} { state := state start := func(kv jetstream.KeyValue) { @@ -115,6 +115,22 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib }) } +// viewsEmitter is the assigner's emitter of its views family: a change not said past views.Bound is its finding +// view-unsaid, in its findings (review round 2); with no findings state it is said in the log only. +func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, findings assign.Findings, tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter { + em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.StreamSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf} + if findings.Put != nil && findings.Delete != nil { + em.Findings = assignerFindings{findings} + } + return em +} + +// assignerFindings is the assigner's findings as views.Emitter's Findings. +type assignerFindings struct{ f assign.Findings } + +func (a assignerFindings) Put(key string, value []byte) error { return a.f.Put(key, value) } +func (a assignerFindings) Delete(key string) error { return a.f.Delete(key) } + // firstViewsBucket is the assigner's own state of its first views (views.StatusState). var firstViewsBucket = delivery.AssignerModule + "_" + views.StatusState 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 4d9c075b..af6761a1 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 @@ -12,6 +12,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-indexer/internal/index" ) @@ -71,6 +72,10 @@ func (m *memStore) Delete(_ context.Context, k string, _ uint64) error { delete( const head = "0123456789abcdef0123456789abcdef01234567" 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 { + t.Error("the indexer's views emitter writes no view-unsaid finding") + } t.Run("not a forge message: terminated and logged", func(t *testing.T) { // 1238 (never redelivered) var logs bytes.Buffer st := &memStore{vals: map[string][]byte{}} 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 0e9e6d28..d7cd5dc8 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -47,8 +47,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun } } marks.Ready(func(mk jetstream.KeyValue) { - em := views.Emitter{Owner: delivery.IndexerModule, Subject: cred.StreamSubject, Publish: views.JetStreamPublish(js), - Marks: mk, TooLarge: tooLarge, Logf: logf} + em := viewsEmitter(ctx, js, cred, mk, findings, tooLarge, logf) for _, state := range sortedStates(states) { states[state](func(kv jetstream.KeyValue) { if err := em.Start(ctx, state, kv, errs); err != nil { @@ -59,6 +58,24 @@ 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 { + w := notHeardWriter(ctx, findings, logf) + return views.Emitter{Owner: delivery.IndexerModule, Subject: cred.StreamSubject, Publish: views.JetStreamPublish(js), + Marks: mk, TooLarge: tooLarge, Findings: funcFindings{put: w.Put, del: w.Delete}, Logf: logf} +} + +// funcFindings is a findings state as two functions: views.Emitter's Findings. +type funcFindings struct { + put func(string, []byte) error + del func(string) error +} + +func (f funcFindings) Put(key string, value []byte) error { return f.put(key, value) } +func (f funcFindings) Delete(key string) error { return f.del(key) } + func sortedStates(states map[string]readyOf) []string { out := make([]string, 0, len(states)) for s := range states {