diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go new file mode 100644 index 00000000..8a628be6 --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/findingwriter.go @@ -0,0 +1,33 @@ +package main + +import ( + "context" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + condseat "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/views" +) + +// findingWriter is the teller's mesh-sdk finding writer (design 47 G §461, §689): the teller observes only what its views +// reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer +// forms its subjects itself (design 47 G §383, A-13). +func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { + w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ + Module: cred.Module, State: state, Node: cred.Node, JS: js, Conn: conn, Logf: logf, + Observe: func(context.Context) (views.Observed[condseat.Finding], error) { + return views.Observed[condseat.Finding]{}, nil + }, + }) + if err != nil { + return nil, err + } + // 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 (as the node-engine's is). + if err := w.Start(ctx); err != nil { + logf("the finding writer cannot start, its findings are kept in the state alone: %v", err) + } + return w, nil +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index 86ba1c0b..636318ee 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -187,6 +187,9 @@ func fullConf(dir string) string { seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>") // The durable consumer on EVENTS the controller makes for the seat it uses, pulled and acknowledged (issue 1126). teller = append(teller, "$JS.API.CONSUMER.INFO.EVENTS."+tellerConsumer, "$JS.API.CONSUMER.MSG.NEXT.EVENTS."+tellerConsumer, "$JS.ACK.EVENTS."+tellerConsumer+".>") + // The finding writer finds the stream of the keeper's findings-asked (design 47 G §384): the bed has none, so it is + // said and the findings stay in the state. + teller = append(teller, "$JS.API.STREAM.NAMES") teller = append(teller, "mesh.mod.mesh-delivery-teller.event.answered") // the manifest's emits (ADR 0596 §2) teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads // The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index a73eb3ba..9ac4db39 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -11,7 +11,6 @@ import ( "context" "errors" "fmt" - sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "io" "log" @@ -104,8 +103,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return calls.Emit(answeredEvent, Answered{Key: key, Ask: askID, Answer: answer}, stdio.EmitOptions{MessageID: answeredID(key, askID)}) } reader := &views.ProcessReader{Module: tellerModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} - findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) - reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) } + writer, err := findingWriter(ctx, cred, conn, js, calls.State("findings"), logger.Printf) + if err != nil { + return fmt.Errorf("the teller's finding writer: %w", err) + } + defer writer.Close() + reader.Findings = func() views.FindingWriter { return views.WriterFindings{Writer: writer} } asks, err := reader.Follow(assignerModule, assignerAsksState, assignerModule+".assigner_read_asks") if err != nil { return err diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index 4f586c1a..a8b768e0 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -29,7 +29,6 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 ) @@ -43,7 +42,7 @@ require ( git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 // indirect - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 // indirect