The verdict events are the seat mesh-catalog's: the indexer says manifest-accepted and manifest-refused on the seat's subject and no longer lists them in emits; builder, watchman and the grants part consume mesh-catalog.<event>, the bus module's keeper hears a watched seat's events (B7d finding 1; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
@@ -235,7 +236,7 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account
|
||||
|
||||
// The events on the builder's consumer of EVENTS that wake a floor (design 51 *The floor*, DLV-B3b).
|
||||
var (
|
||||
manifestAcceptedSubject = "mesh.mod." + delivery.IndexerModule + ".event." + delivery.ManifestAcceptedEvent
|
||||
manifestAcceptedSubject = meshcatalog.SubjectManifestAccepted
|
||||
conditionClearedSubject = "mesh.mod." + health.KeeperModule + ".event." + health.EventCleared
|
||||
)
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ module mesh-delivery-builder
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
|
||||
|
||||
@@ -134,9 +134,9 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
// DLV-M1: the indexer's and the assigner's views families of the states it follows.
|
||||
if want := []string{"container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed",
|
||||
"go-package-registry.kept", "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed",
|
||||
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
|
||||
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted",
|
||||
"mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.placement", "mesh-delivery-checker.tag-wanted", "mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery-indexer.manifest-accepted", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-indexer.views.packages", "node-engine.manifest-reads-changed",
|
||||
"npm-package-registry.kept", "npm-package-registry.published", "npm-package-registry.refused", "npm-package-registry.removed",
|
||||
"oci-artifact-registry.kept", "oci-artifact-registry.published", "oci-artifact-registry.refused", "oci-artifact-registry.removed"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
|
||||
@@ -39,10 +39,10 @@
|
||||
"mesh-artifact-sweep.artifacts-gone",
|
||||
"mesh-build-go.toolchain.reported",
|
||||
"mesh-build-node.toolchain.reported",
|
||||
"mesh-catalog.manifest-accepted",
|
||||
"mesh-conditions.condition-cleared",
|
||||
"mesh-delivery-checker.tag-wanted",
|
||||
"mesh-delivery-forge-bridge.forge.**",
|
||||
"mesh-delivery-indexer.manifest-accepted",
|
||||
"node-engine.manifest-reads-changed",
|
||||
"npm-package-registry.kept",
|
||||
"npm-package-registry.published",
|
||||
|
||||
@@ -534,13 +534,13 @@ func startIndexerRun(t *testing.T, put func(context.Context, buildLog), before .
|
||||
t.Fatal(err)
|
||||
}
|
||||
acceptedHeard.Store(0)
|
||||
waiter, err := ctl.Subscribe("mesh.mod.mesh-delivery-indexer.event.manifest-accepted", func(*nats.Msg) { acceptedHeard.Add(1) })
|
||||
waiter, err := ctl.Subscribe("mesh.seat.mesh-catalog.event.manifest-accepted", func(*nats.Msg) { acceptedHeard.Add(1) })
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = waiter.Unsubscribe() })
|
||||
// The shared events stream: the indexer's ordinary events and the assigner's views events (ADR 0635 §5).
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
put(ctx, buildLog{ctx: ctx, js: js})
|
||||
@@ -733,7 +733,7 @@ func TestIndexerKeepsManifestsAtAMerge(t *testing.T) { // 1444/AC4, 1449/AC4, 15
|
||||
break
|
||||
}
|
||||
var r delivery.ManifestRefused
|
||||
if msg.Subject == "mesh.mod.mesh-delivery-indexer.event."+delivery.ManifestRefusedEvent && json.Unmarshal(msg.Data, &r) == nil {
|
||||
if msg.Subject == "mesh.seat.mesh-catalog.event."+delivery.ManifestRefusedEvent && json.Unmarshal(msg.Data, &r) == nil {
|
||||
said[r.Module] = true
|
||||
}
|
||||
}
|
||||
|
||||
+8
-7
@@ -78,14 +78,10 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
|
||||
}
|
||||
}
|
||||
serveForge(t, js.Conn(), listing, goSumContent)
|
||||
// The seat definitions' state and stream (DLV-I17), as the grants part makes them from the manifest.
|
||||
// The seat definitions' state (DLV-I17), as the grants part makes it from the manifest.
|
||||
if seatDefs, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.SeatDefinitionsBucket, History: 5}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_MESH_CATALOG",
|
||||
Subjects: []string{"mesh.seat." + index.CatalogSeat + ".event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// DLV-I19, I20: the indexer's findings and takeins states, and the checker's stream asks with the indexer's durable
|
||||
// consumer, which carries the checker's verdict event.
|
||||
if findings, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: findingsBucket, History: 1}); err != nil {
|
||||
@@ -263,12 +259,17 @@ func TestAMeshSDKMergeMovesOnlyTheModulesImportingAChangedPackage(t *testing.T)
|
||||
t.Fatalf("%s is kept 20 s after its pull request closed", pk)
|
||||
}
|
||||
}
|
||||
info, err := js.Stream(ctx, "SEAT_MESH_CATALOG")
|
||||
// The seat's events are on the shared events stream like every seat's.
|
||||
info, err := js.Stream(ctx, "EVENTS")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
full, err := info.Info(ctx, jetstream.WithSubjectFilter("mesh.seat."+index.CatalogSeat+".event.seat-definition-changed"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Said on the seat's subject: the trunk's git alone; a pull request's head (unparsed, then closed) says nothing (ADR 0636 §4).
|
||||
if n := info.CachedInfo().State.Msgs; n != 1 {
|
||||
if n := full.State.Subjects["mesh.seat."+index.CatalogSeat+".event.seat-definition-changed"]; n != 1 {
|
||||
t.Errorf("seat-definition-changed said %d times, want 1", n)
|
||||
}
|
||||
if m, err := info.GetLastMsgForSubject(ctx, "mesh.seat.mesh-catalog.event.seat-definition-changed"); err != nil ||
|
||||
|
||||
@@ -13,7 +13,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// sayRefused says the indexer's event manifest-refused (novox/hq ADR 0578 §1): mesh.mod.<module>.event.manifest-refused,
|
||||
// sayRefused says the indexer's event manifest-refused (novox/hq ADR 0578 §1): mesh.seat.mesh-catalog.event.manifest-refused,
|
||||
// with the SDK's envelope headers and the digest and judge as the bus's dedup id, so one refusal is said once.
|
||||
func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string, ...any)) func(context.Context, delivery.ManifestRefused) {
|
||||
return func(ctx context.Context, r delivery.ManifestRefused) {
|
||||
@@ -22,7 +22,7 @@ func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string,
|
||||
logf("judge: %s for %s is not said: %v", delivery.ManifestRefusedEvent, r.Module, err)
|
||||
return
|
||||
}
|
||||
msg := nats.NewMsg(cred.EventSubject(delivery.ManifestRefusedEvent))
|
||||
msg := nats.NewMsg(seatEventSubject(delivery.ManifestRefusedEvent))
|
||||
id := "manifest-refused-" + r.Digest + "-" + r.Judge
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", cred.Module)
|
||||
@@ -39,7 +39,7 @@ func sayRefused(js jetstream.JetStream, cred process.Account, logf func(string,
|
||||
}
|
||||
|
||||
// sayAccepted says the indexer's event manifest-accepted (issue 1966 D3, novox/hq ADR 0595 §4):
|
||||
// mesh.mod.<module>.event.manifest-accepted, carrying the module's bus-state entries whole, with the digest and judge as
|
||||
// mesh.seat.mesh-catalog.event.manifest-accepted, carrying the module's bus-state entries whole, with the digest and judge as
|
||||
// the bus's dedup id (delivery.ManifestAccepted.EventID), so one acceptance is said once.
|
||||
func sayAccepted(js jetstream.JetStream, cred process.Account) func(context.Context, delivery.ManifestAccepted, string) error {
|
||||
return func(ctx context.Context, a delivery.ManifestAccepted, id string) error {
|
||||
@@ -47,7 +47,7 @@ func sayAccepted(js jetstream.JetStream, cred process.Account) func(context.Cont
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := nats.NewMsg(cred.EventSubject(delivery.ManifestAcceptedEvent))
|
||||
msg := nats.NewMsg(seatEventSubject(delivery.ManifestAcceptedEvent))
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", cred.Module)
|
||||
msg.Header.Set("x-node", cred.Node)
|
||||
|
||||
@@ -3,6 +3,7 @@ module mesh-delivery-indexer
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
|
||||
|
||||
@@ -4355,7 +4355,7 @@ func TestTheBussAccountSaysNothingOnTheBus(t *testing.T) {
|
||||
keeper := func(bus string, more string) []byte {
|
||||
return []byte(`{"module":"nats","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],
|
||||
"claims":[{"name":"mesh-broker","scope":"mesh","serves":["list"]}],"own-secrets":{"broker":"/run/broker"},
|
||||
"tools":["nats_server"],"consumes":["mesh-delivery-indexer.manifest-accepted"],"uses":["operator-channel"],
|
||||
"tools":["nats_server"],"consumes":["mesh-catalog.manifest-accepted"],"uses":["operator-channel"],
|
||||
"state":[{"name":"requests","history":5}],"streams":[{"name":"broker-events","events":["created"]}],
|
||||
"emits":["consumer-removed"],
|
||||
"resources":[{"id":"nats-keeper","type":"user","name":"nats-keeper"},
|
||||
|
||||
@@ -109,9 +109,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
}
|
||||
// 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state. ADR 0578 §1: it says
|
||||
// a refused manifest as manifest-refused, and (issue 1966 D3, 1966/AC10) a newly accepted one asking for bus state as
|
||||
// manifest-accepted.
|
||||
if own := slices.DeleteFunc(slices.Clone(m.Emits), func(e string) bool { return strings.HasPrefix(e, views.EventPrefix) }); !reflect.DeepEqual(own, []string{"manifest-refused", "manifest-accepted", takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
|
||||
t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits)
|
||||
// manifest-accepted, both events of the seat mesh-catalog it holds (ADR 0636 §4), so not among its `emits`.
|
||||
if own := slices.DeleteFunc(slices.Clone(m.Emits), func(e string) bool { return strings.HasPrefix(e, views.EventPrefix) }); !reflect.DeepEqual(own, []string{takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
|
||||
t.Errorf("emits = %v, want the two take-in events (the verdict events are the seat mesh-catalog's)", m.Emits)
|
||||
}
|
||||
// 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: "modules", History: 20}, {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: delivery.PackagesState, History: delivery.PackagesHistory},
|
||||
|
||||
@@ -89,8 +89,6 @@
|
||||
{ "name": "pending", "history": 1 }
|
||||
],
|
||||
"emits": [
|
||||
"manifest-refused",
|
||||
"manifest-accepted",
|
||||
"take-in.pending",
|
||||
"take-in.taken",
|
||||
"views.candidates",
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
@@ -16,7 +17,7 @@ import (
|
||||
"mesh-delivery-watchman/internal/watch"
|
||||
)
|
||||
|
||||
// **What the watchman consumes** (its manifest's `consumes`; novox/hq ADR 0595 §4, design 25 § 12.2; DLV-W1): the indexer's
|
||||
// **What the watchman consumes** (its manifest's `consumes`; novox/hq ADR 0595 §4, design 25 § 12.2; DLV-W1): the seat mesh-catalog's
|
||||
// manifest-accepted and the seat mesh-broker's created and refused, all on its durable consumer on EVENTS (the seat's
|
||||
// events on the seat's subjects like every seat's: the lead's ruling, registry milestone 13:0x); beside it any module
|
||||
// stream its membership names (Hears). Each consumer is the one the grants part makes; one
|
||||
@@ -34,7 +35,7 @@ const (
|
||||
// and refused. Anything else is left.
|
||||
func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, ...any)) {
|
||||
switch {
|
||||
case strings.HasSuffix(subject, ".event."+eventAccepted) && strings.HasPrefix(subject, "mesh.mod."):
|
||||
case subject == meshcatalog.SubjectManifestAccepted:
|
||||
var a delivery.ManifestAccepted
|
||||
if err := json.Unmarshal(body, &a); err != nil || a.Module == "" {
|
||||
logf("a manifest-accepted that does not read is left: %v", err)
|
||||
|
||||
@@ -64,7 +64,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
|
||||
if !out.findings[watch.FindingKey(watch.ScopeModule, "m", watch.KindBusStateRefused+"-l")] {
|
||||
t.Fatalf("refused reached no finding: %v", out.findings)
|
||||
}
|
||||
consumed(w, "mesh.mod.mesh-delivery-indexer.event.manifest-accepted", []byte(`{"module":"m","digest":"d","bus-state":[]}`), say)
|
||||
consumed(w, "mesh.seat.mesh-catalog.event.manifest-accepted", []byte(`{"module":"m","digest":"d","bus-state":[]}`), say)
|
||||
if len(out.findings) != 0 {
|
||||
t.Fatalf("an accepted manifest no longer asking for l left %v", out.findings)
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ module mesh-delivery-watchman
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
|
||||
|
||||
@@ -109,7 +109,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably.
|
||||
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.manifest-accepted", "mesh-broker.created", "mesh-broker.refused"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "mesh-broker.created", "mesh-broker.refused"}; !reflect.DeepEqual(m.Consumes, want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
|
||||
|
||||
@@ -35,7 +35,7 @@
|
||||
"mesh-delivery-indexer.take-in.pending",
|
||||
"mesh-delivery-indexer.take-in.taken",
|
||||
"mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-indexer.manifest-accepted",
|
||||
"mesh-catalog.manifest-accepted",
|
||||
"mesh-broker.created",
|
||||
"mesh-broker.refused"
|
||||
],
|
||||
|
||||
@@ -32,11 +32,11 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
"mesh-grants/internal/grants"
|
||||
@@ -157,9 +157,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
hearViews(m)
|
||||
return
|
||||
}
|
||||
// The indexer's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision.
|
||||
if event, ok := strings.CutPrefix(m.Subject(), "mesh.mod."+grants.IndexerModule+".event."); ok &&
|
||||
(event == delivery.ManifestAcceptedEvent || event == delivery.ManifestRefusedEvent) {
|
||||
// The seat mesh-catalog's verdict events go to the verdicts' first view (ADR 0622 §2), which takes each at its revision.
|
||||
if event, ok := strings.CutPrefix(m.Subject(), "mesh.seat."+meshcatalog.Seat+".event."); ok &&
|
||||
(event == meshcatalog.EventManifestAccepted || event == meshcatalog.EventManifestRefused) {
|
||||
if err := fv.HearVerdict(ctx, event, m.Data(), logger.Printf); err != nil {
|
||||
logger.Printf("views: %s is not a verdict event, dropped: %v", m.Subject(), err)
|
||||
_ = m.Term()
|
||||
|
||||
@@ -57,7 +57,7 @@
|
||||
"mesh-delivery-indexer.indexer_bus_state"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-indexer.manifest-accepted"
|
||||
"mesh-catalog.manifest-accepted"
|
||||
],
|
||||
"uses": [
|
||||
"operator-channel"
|
||||
|
||||
@@ -3,6 +3,7 @@ module mesh-grants
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshnodeenroller v0.0.0-00010101000000-000000000000
|
||||
|
||||
@@ -1097,7 +1097,7 @@ func theBusModulesProcessKeepsBusState(t *testing.T) {
|
||||
{Module: "node-tools", ToolRunner: true},
|
||||
{Module: "nats", SnapshotsTheBus: true, Holds: []Seat{{Name: meshbroker.Seat, Serves: BrokerVerbs(),
|
||||
Emits: []string{"created", "refused"}}},
|
||||
Consumes: []string{"mesh-delivery-indexer.manifest-accepted"}, Uses: []Seat{{Name: "operator-channel", Accepts: []string{"condition"}, Emits: []string{"decided"}, ByCaller: []string{"condition", "decided"}}},
|
||||
Watches: []Seat{{Name: "mesh-catalog", Emits: []string{"manifest-accepted"}}}, Uses: []Seat{{Name: "operator-channel", Accepts: []string{"condition"}, Emits: []string{"decided"}, ByCaller: []string{"condition", "decided"}}},
|
||||
Streams: []ModuleStream{{Module: "nats", Name: "BROKER_EVENTS", Events: []string{"created", "refused"}, MaxAgeDays: 30, MaxMiB: 256}},
|
||||
State: []Bucket{{Module: "nats", Name: "spent", ReadOnly: true}},
|
||||
Processes: []ProcessPrincipal{{Process: "keeper", State: []string{"asked", "spent"}, Keeps: true}, {Process: "other"},
|
||||
@@ -1214,7 +1214,7 @@ func theBusModulesProcessKeepsBusState(t *testing.T) {
|
||||
made := false
|
||||
for _, mc := range AccountConsumers(users) {
|
||||
if mc.Consumer.Stream == "EVENTS" && mc.Consumer.Name == "a_nats_keeper" &&
|
||||
slices.Equal(mc.Consumer.Filters, []string{"mesh.mod.mesh-delivery-indexer.event.manifest-accepted",
|
||||
slices.Equal(mc.Consumer.Filters, []string{"mesh.seat.mesh-catalog.event.manifest-accepted",
|
||||
"mesh.seat.operator-channel.event.decided.nats"}) {
|
||||
made = true
|
||||
}
|
||||
|
||||
@@ -93,6 +93,19 @@ func keeperConsumer(p Principal) (Consumer, bool) {
|
||||
filters = append(filters, s)
|
||||
}
|
||||
}
|
||||
// And the events of the seats its module watches (its consumes naming a seat's event, such as the seat mesh-catalog's
|
||||
// manifest-accepted), on the seat's own subjects.
|
||||
for _, w := range p.Watches {
|
||||
if w.Kinded {
|
||||
continue
|
||||
}
|
||||
for _, e := range w.Emits {
|
||||
if pattern, ok := machineEvent(w, e); ok {
|
||||
e = pattern
|
||||
}
|
||||
filters = append(filters, seatSubject(w, "event", e))
|
||||
}
|
||||
}
|
||||
// And the events of the seats its module uses that are said to it by name, the operator channel's `decided` for its own
|
||||
// asks among them (design 25 § 12.3: the process wakes on the warrant, heard on its durable consumer, never on a timer).
|
||||
for _, s := range SeatTrafficOf(p.Module, nil, p.Uses, nil).Subscribe {
|
||||
|
||||
@@ -249,6 +249,7 @@ func Users(r Records) ([]Principal, error) {
|
||||
k := &out[len(out)-1]
|
||||
k.KeepsBus, k.KeepsBusState = true, BusStateNames(r.BusState, r.Bound)
|
||||
k.Consumes, k.Streams = d.Consumes, d.Streams
|
||||
k.Watches = d.Watches
|
||||
k.SaysSeatEvents = brokerSeatEvents(d.Holds)
|
||||
}
|
||||
if d.SnapshotsTheBus && len(pr.Removes) > 0 {
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
|
||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
@@ -22,7 +23,7 @@ import (
|
||||
// DLV-R1). The part reads the assigner's placement and wants and the indexer's manifests, modules and verdicts no longer
|
||||
// from their buckets, but by each owner's first-view tool, asked once at its start, then from the owner's views family
|
||||
// alone (each event one changed key with its revision), asking once more only on a gap in revisions, never on a timer.
|
||||
// The verdicts alone are followed on the indexer's manifest-accepted and manifest-refused, which carry the revision
|
||||
// The verdicts alone are followed on the seat mesh-catalog's manifest-accepted and manifest-refused, which carry the revision
|
||||
// (ADR 0622 §2), not on its views family (HearVerdict).
|
||||
// Started while an owner is down it keeps asking and never reads the owner's state; past the bound (two minutes) it
|
||||
// raises FirstViewUnanswered naming the owner and the tool, cleared by the answer.
|
||||
@@ -260,11 +261,11 @@ func (v *Views) Again(ctx context.Context, state, why string) ([]string, error)
|
||||
return asked, nil
|
||||
}
|
||||
|
||||
// VerdictEvents are the indexer's two events the part follows the verdicts on after their first view (ADR 0622 §2), as its
|
||||
// VerdictEvents are the seat mesh-catalog's two events the part follows the verdicts on after their first view (ADR 0622 §2), as its
|
||||
// manifest's consumes name them.
|
||||
var VerdictEvents = []string{IndexerModule + "." + delivery.ManifestAcceptedEvent, IndexerModule + "." + delivery.ManifestRefusedEvent}
|
||||
var VerdictEvents = []string{meshcatalog.Seat + "." + meshcatalog.EventManifestAccepted, meshcatalog.Seat + "." + meshcatalog.EventManifestRefused}
|
||||
|
||||
// HearVerdict takes one of the indexer's manifest-accepted or manifest-refused (event, its name) into the verdicts' first
|
||||
// HearVerdict takes one of the seat mesh-catalog's manifest-accepted or manifest-refused (event, its name) into the verdicts' first
|
||||
// view, as one change of the verdicts state at the revision it carries (ADR 0622 §2): the manifest's digest its key, the
|
||||
// verdict it says its value. A revision past the one after the picture's is a gap, and so is an event carrying no
|
||||
// revision (an indexer that does not say it) or one saying no manifest above the picture's revision (the empty accepted
|
||||
|
||||
@@ -21,8 +21,8 @@
|
||||
"mesh-delivery-assigner.views.wants",
|
||||
"mesh-delivery-indexer.views.manifests",
|
||||
"mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-indexer.manifest-accepted",
|
||||
"mesh-delivery-indexer.manifest-refused"
|
||||
"mesh-catalog.manifest-accepted",
|
||||
"mesh-catalog.manifest-refused"
|
||||
],
|
||||
"uses": [
|
||||
"operator-channel"
|
||||
|
||||
Reference in New Issue
Block a user