CAT-RD4: mesh-conditions asks the assigner's wants, placement and findings once and follows their events, no reads (on CAT-C10's base)
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Executable
BIN
Binary file not shown.
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user