diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index a70504ec..304652ba 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -8,12 +8,16 @@ package main import ( + "context" "fmt" "os" "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-delivery/sdk/views" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -59,10 +63,22 @@ func NewKeeper() *Keeper { // watch keeps a watch of one state, said blind while it cannot be read: a state not on the bus yet (an observer not // delivered) is read again when it is. -func watch(k *Keeper, state, observer string, handle func(stdio.StateChange) error) { +func watch(k *Keeper, v *views.Reader, state, observer string, handle func(stdio.StateChange) error) { go func() { for wait := 5 * time.Second; ; wait = min(wait*2, 5*time.Minute) { - err := stdio.State(state).Watch("", handle) + var err error + if v.Follows(state) { + // The assigner's states are asked once of its read tool and followed by its events (ADR 0615). + err = v.Watch(state, func(key string, value []byte, gone bool) { + op := stdio.OpPut + if gone { + op = stdio.OpDelete + } + _ = handle(stdio.StateChange{Key: key, Value: value, Op: op}) + }) + } else { + err = stdio.State(state).Watch("", handle) + } if err == nil { if observer != "" { k.Blind(observer, "") @@ -79,7 +95,7 @@ func watch(k *Keeper, state, observer string, handle func(stdio.StateChange) err } // Keep starts every watch the keeper reads by (clauses 2, 8, 9, 13). -func Keep(k *Keeper) { +func Keep(k *Keeper, v *views.Reader) { for _, s := range []string{HeldState} { keys, _ := stdio.State(s).Keys() for _, key := range keys { @@ -99,35 +115,62 @@ func Keep(k *Keeper) { if n, err := k.EndCarried(); n > 0 || err != nil { logf("cleared %d condition(s) of kinds that ended with the controller: %v", n, err) } - watch(k, WantsRead, "", func(c stdio.StateChange) error { return said(k.Want(c.Key, c.Value, c.Op != "put")) }) - watch(k, HavesRead, "", func(c stdio.StateChange) error { return said(k.Have(c.Key, c.Value, c.Op != "put")) }) - watch(k, AttemptsRead, health.HealersModule+" attempts", func(c stdio.StateChange) error { + watch(k, v, WantsRead, "", func(c stdio.StateChange) error { return said(k.Want(c.Key, c.Value, c.Op != "put")) }) + watch(k, v, HavesRead, "", func(c stdio.StateChange) error { return said(k.Have(c.Key, c.Value, c.Op != "put")) }) + watch(k, v, AttemptsRead, health.HealersModule+" attempts", func(c stdio.StateChange) error { return said(k.Attempt(c.Key, c.Value, c.Op != "put")) }) for _, o := range Observers { - watch(k, o+"."+health.FindingsState, o, func(c stdio.StateChange) error { + watch(k, v, o+"."+health.FindingsState, o, func(c stdio.StateChange) error { return said(k.Finding(o, c.Key, c.Value, c.Op != "put")) }) } - watch(k, BoundsState, "", func(c stdio.StateChange) error { + watch(k, v, BoundsState, "", func(c stdio.StateChange) error { if c.Op == stdio.OpExpire { return said(k.BoundPassed(c.Key)) } return nil }) - watch(k, PlacementRead, "", func(c stdio.StateChange) error { + watch(k, v, PlacementRead, "", func(c stdio.StateChange) error { k.Placement(c.Key, c.Value, c.Op != stdio.OpPut) return nil }) } +// The assigner's read tools of the states the keeper follows, which the manifest invokes (ADR 0615). +const ( + WantsTool = delivery.AssignerModule + ".assigner_read_wants" + PlacementTool = delivery.AssignerModule + ".assigner_read_placement" + FindingsTool = delivery.AssignerModule + ".assigner_read_findings" +) + +// NewViews is the keeper's pictures of the assigner's wants, placement and findings. A first view it cannot get past its +// bound is a finding of its own, taken in as any observer's: the keeper raises the condition itself. +func NewViews(k *Keeper) *views.Reader { + v := &views.Reader{Module: Module, Logf: logf, Findings: ownFindings{k}} + v.Follow(delivery.AssignerModule, "wants", WantsTool) + v.Follow(delivery.AssignerModule, "placement", PlacementTool) + v.Follow(delivery.AssignerModule, health.FindingsState, FindingsTool) + return v +} + +// ownFindings takes the keeper's own finding straight into its picture of findings. +type ownFindings struct{ k *Keeper } + +func (o ownFindings) Put(key string, value []byte) error { + return o.k.Finding(Module, key, value, false) +} +func (o ownFindings) Delete(key string) error { return o.k.Finding(Module, key, nil, true) } + func main() { k := NewKeeper() + v := NewViews(k) go func() { <-stdio.Served() - Keep(k) + v.Start(context.Background()) + Keep(k, v) }() - if err := stdio.Serve("", Tools(k)); err != nil { + if err := stdio.Serve("", append(Tools(k), v.Tools("mesh_conditions")...)); err != nil { logf("%v", err) os.Exit(1) } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 79ccbe5c..74cba26e 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -7,6 +7,7 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -29,9 +30,17 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { !slices.Equal(Verbs, catalogue.VerbNames(seat.Serves)) { t.Errorf("module %s runs as %s and claims %+v; the seat serves %v", m.Module, m.RunsAs, m.Claims, catalogue.VerbNames(seat.Serves)) } - reads := []string{WantsRead, HavesRead, AttemptsRead, PlacementRead} + // The assigner's wants, placement and findings are asked once of its read tools and followed by its events, not read + // (ADR 0615 ยง1). + reads := []string{HavesRead, AttemptsRead} for _, o := range Observers { - reads = append(reads, o+"."+health.FindingsState) + if o != delivery.AssignerModule { + reads = append(reads, o+"."+health.FindingsState) + } + } + if !slices.Equal(m.Invokes, []string{WantsTool, PlacementTool, FindingsTool}) || + !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.findings"}) { + t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } slices.Sort(reads) got := slices.Clone(m.Reads) diff --git a/modules/mesh-conditions/mesh-conditions b/modules/mesh-conditions/mesh-conditions new file mode 100755 index 00000000..36d5a3bc Binary files /dev/null and b/modules/mesh-conditions/mesh-conditions differ diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index e3ce5c36..45608010 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -17,15 +17,22 @@ "mesh-delivery-watchman.findings", "mesh-self-check.findings", "mesh-healers.findings", - "mesh-delivery-assigner.findings", "messenger.findings", "mesh-provisioner.findings", "artifact-store-tools.findings", "mesh-config.findings", "mesh-healers.attempts", - "mesh-delivery-assigner.wants", - "node-engine.have", - "mesh-delivery-assigner.placement" + "node-engine.have" + ], + "invokes": [ + "mesh-delivery-assigner.assigner_read_wants", + "mesh-delivery-assigner.assigner_read_placement", + "mesh-delivery-assigner.assigner_read_findings" + ], + "consumes": [ + "mesh-delivery-assigner.views.wants", + "mesh-delivery-assigner.views.placement", + "mesh-delivery-assigner.views.findings" ], "state": [ "conditions",