The view says modules in motion: moving, settled and stuck, once for a node, a module and a composition (issue 1906 D10, for the messenger)

This commit is contained in:
2026-10-24 18:27:01 +02:00
parent 8280506cee
commit a3815051cc
6 changed files with 455 additions and 5 deletions
@@ -791,3 +791,135 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
t.Errorf("the headers = %v", msg.Header)
}
}
// 1906/D10, whole on the bus (through run): a want that differs from its have is said `moving`, the have reaching it running and
// healthy `settled`, a have held or failed `stuck`, each once with the body the messenger reads and an event id per type, node, module
// and composition.
func TestProcessWholeSaysModulesInMotion(t *testing.T) {
srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("the bus did not come up")
}
t.Cleanup(srv.Shutdown)
nc, err := nats.Connect(srv.ClientURL())
if err != nil {
t.Fatal(err)
}
t.Cleanup(nc.Close)
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}})
if err != nil {
t.Fatal(err)
}
kvs := map[string]jetstream.KeyValue{}
for _, name := range []string{delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.StuckBucket} {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
if err != nil {
t.Fatal(err)
}
kvs[name] = kv
}
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
broker := filepath.Join(dir, "broker")
cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"}
runCtx, stop := context.WithCancel(ctx)
done := make(chan error, 1)
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
t.Cleanup(func() { stop(); <-done })
time.Sleep(2 * time.Second) // the view has read what stood: nothing
fpOld, fpNew := "src2:"+strings.Repeat("1", 64), "src2:"+strings.Repeat("2", 64)
digest := "sha256:" + strings.Repeat("c", 64)
put := func(bucket, key string, v []byte, err error) {
t.Helper()
if err != nil {
t.Fatal(err)
}
if _, err := kvs[bucket].Put(ctx, key, v); err != nil {
t.Fatal(err)
}
}
want := func(node, module string) {
v, err := delivery.EncodeWant(delivery.Want{FP: fpNew, Composed: digest})
put(delivery.AssignerWantsBucket, "want."+node+"."+module, v, err)
}
have := func(node, module, fp string, phase delivery.Phase, health delivery.Health, detail string) {
v, err := delivery.EncodeHave(delivery.Have{FP: fp, Phase: phase, Health: health, Detail: detail})
put(delivery.HaveBucket, node+"."+module, v, err)
}
have("ace", "web", fpOld, delivery.PhaseRunning, delivery.HealthOK, "")
want("ace", "web")
have("g14", "web", fpOld, delivery.PhaseRunning, delivery.HealthOK, "")
want("g14", "web")
said := func() map[string]jetstream.RawStreamMsg {
out := map[string]jetstream.RawStreamMsg{}
for seq := uint64(1); ; seq++ {
m, err := events.GetMsg(ctx, seq)
if err != nil {
return out
}
out[m.Header.Get("x-event-id")] = *m
}
}
waitFor := func(id string) jetstream.RawStreamMsg {
t.Helper()
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
if m, ok := said()[id]; ok {
return m
}
}
t.Fatalf("never said %s; said %v", id, func() []string {
var ids []string
for k := range said() {
ids = append(ids, k)
}
return ids
}())
return jetstream.RawStreamMsg{}
}
hex := strings.Repeat("c", 64)
mv := waitFor("moving.ace.web." + hex)
var body view.MotionBody
if err := json.Unmarshal(mv.Data, &body); err != nil || body.Node != "ace" || body.Module != "web" || body.FromFP != fpOld || body.ToFP != fpNew || body.Composed != digest || body.BoundSeconds != view.MotionBoundSeconds || body.Since.IsZero() {
t.Errorf("moving = %+v, %v", body, err)
}
if mv.Subject != "mesh.mod.mesh-delivery-view.event.moving" {
t.Errorf("subject = %s", mv.Subject)
}
have("ace", "web", fpNew, delivery.PhaseRunning, delivery.HealthOK, "")
st := waitFor("settled.ace.web." + hex)
if err := json.Unmarshal(st.Data, &body); err != nil || body.Outcome != "running" {
t.Errorf("settled = %+v, %v", body, err)
}
have("g14", "web", fpOld, delivery.PhaseHeld, delivery.HealthWaits, "no space left")
stk := waitFor("stuck.g14.web." + hex)
if err := json.Unmarshal(stk.Data, &body); err != nil || body.Outcome != "held" || body.Words != "no space left" {
t.Errorf("stuck = %+v, %v", body, err)
}
time.Sleep(time.Second)
count := map[string]int{}
for id := range said() {
count[strings.SplitN(id, ".", 2)[0]]++
}
if count["moving"] != 2 || count["settled"] != 1 || count["stuck"] != 1 {
t.Errorf("events said by type = %v, want moving 2, settled 1, stuck 1", count)
}
}
@@ -138,6 +138,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
model.SetWriter(kvWriter{ctx: ctx, js: js})
model.SetMotionEvent(func(e view.MotionEvent) {
if err := emitEvent(ctx, js, cred, e.Type, e.Body, e.ID); err != nil {
logger.Printf("the %s of %s on %s could not be said: %v", e.Type, e.Body.Module, e.Body.Node, err)
}
})
model.SetDeliveredEvent(func(e view.DeliveredEvent) {
if err := emitEvent(ctx, js, cred, DeliveredEventType, e, fmt.Sprintf("delivered.%d", e.Revision)); err != nil {
logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err)
@@ -209,6 +214,7 @@ func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model
return errors.New("the watch ended")
}
if e == nil { // the last values have all been delivered
model.Initial(bucket)
continue
}
model.Apply(view.Entry{Bucket: bucket, Key: e.Key(), Value: e.Value(), At: e.Created(),
@@ -222,6 +222,9 @@ type Model struct {
wants map[string]wantEntry // the assigner's wants, by <node>.<module>
source *sourceEntry // the gate's source key
deliveredEvent func(DeliveredEvent)
motionEvent func(MotionEvent)
told map[string]bool // the motion events said, by their ids
initial map[string]bool // the buckets whose watch handed over what stood at its start
said uint64 // the source revision whose delivery was said
composed map[string]composedEntry // the controller's digests and judgements, by <node>.<module>.<hex>
aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key
@@ -297,6 +300,7 @@ type keyEntry struct {
type wantEntry struct {
value delivery.Want
err string
at time.Time
}
type composedEntry struct {
@@ -340,7 +344,7 @@ type nodeEntry struct {
// New is an empty model.
func New() *Model {
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, newest: map[string]string{},
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
}
@@ -352,10 +356,16 @@ func (m *Model) Apply(e Entry) {
m.apply(e)
event := m.deliveredNow()
say := m.deliveredEvent
motion, sayMotion := m.motionNow(e), m.motionEvent
m.mu.Unlock()
if event != nil && say != nil {
say(*event)
}
if sayMotion != nil {
for _, ev := range motion {
sayMotion(ev)
}
}
}
// DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR
@@ -489,7 +499,7 @@ func (m *Model) apply(e Entry) {
return
}
w, err := delivery.DecodeWant(e.Value)
m.wants[key] = wantEntry{value: w, err: errText(err)}
m.wants[key] = wantEntry{value: w, err: errText(err), at: e.At}
case delivery.ControllerComposedBucket:
if e.Deleted {
delete(m.composed, e.Key)
@@ -1086,3 +1096,183 @@ func splitKey(key string) (node, module string) {
}
return node, module
}
// The events of a module in motion (issue 1906 D10, for the messenger, ADR 0308): with no plan to say it, the view says that a
// module's want on a node differs from its have (moving), that the have reached the want running and healthy (settled), or that it
// is held or failed or stopped moving past the watchman's bound (stuck). Each is said once for a node, a module and a composition.
const (
MotionMoving = "moving"
MotionSettled = "settled"
MotionStuck = "stuck"
// MotionBoundSeconds is how long a node may take to apply a want before the watchman says it stuck (watch.StepBound).
MotionBoundSeconds = 1800
)
// MotionBody is the body of the three events. Outcome and Words are those of settled and stuck: settled is "running"; stuck is
// "held" or "failed" (the have's own words in Words) or "past-bound" (the watchman's words of what it waited for).
type MotionBody struct {
Node string `json:"node"`
Module string `json:"module"`
FromFP string `json:"from_fp"`
ToFP string `json:"to_fp"`
Composed string `json:"composed"`
Since time.Time `json:"since"`
BoundSeconds int `json:"bound_seconds"`
Outcome string `json:"outcome,omitempty"`
Words string `json:"words,omitempty"`
}
// MotionEvent is an event to say: its type (moving, settled or stuck), its id (the bus's dedup id and the event id header, one for
// a type, a node, a module and a composition) and its body.
type MotionEvent struct {
Type string
ID string
Body MotionBody
}
// SetMotionEvent sets where the events of modules in motion go; nothing is said until it is set.
func (m *Model) SetMotionEvent(say func(MotionEvent)) {
m.mu.Lock()
defer m.mu.Unlock()
m.motionEvent = say
}
// motionNow is the events to say now about the node and module the change touched: a want, a have, or the watchman's stuck key of a
// want.
func (m *Model) motionNow(e Entry) []MotionEvent {
if m.motionEvent == nil || !m.motionReady() {
return nil
}
var hk string
switch e.Bucket {
case delivery.AssignerWantsBucket:
node, module, err := delivery.ParseWantKey(e.Key)
if err != nil {
return nil
}
hk, _ = delivery.HaveKey(node, module)
case delivery.HaveBucket:
if _, _, isNode, err := delivery.ParseHaveKey(e.Key); err != nil || isNode {
return nil
}
hk = e.Key
case delivery.StuckBucket:
st, ok := m.stuck[e.Key]
if !ok || st.err != "" || st.value.Bucket != delivery.AssignerWantsBucket {
return nil
}
node, module, err := delivery.ParseWantKey(st.value.Key)
if err != nil {
return nil
}
hk, _ = delivery.HaveKey(node, module)
default:
return nil
}
return m.motionOf(hk)
}
// motionReady says whether the view has read what stands in the buckets motion is derived from (the wants, the haves and the
// watchman's stuck keys): before, a want read ahead of its have would be said moving though the node runs it.
func (m *Model) motionReady() bool {
for _, b := range []string{delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.StuckBucket} {
if !m.initial[b] {
return false
}
}
return true
}
// Initial tells the model that the watch of a bucket has handed over every key that stood at its start. When the buckets motion is
// derived from have all done so, the events of the modules in motion at that moment are said (the ids deduplicate what a restart
// of the view says again).
func (m *Model) Initial(bucket string) {
m.mu.Lock()
was := m.motionReady()
m.initial[bucket] = true
var out []MotionEvent
if !was && m.motionReady() && m.motionEvent != nil {
keys := make([]string, 0, len(m.wants))
for k := range m.wants {
keys = append(keys, k)
}
sort.Strings(keys)
for _, k := range keys {
out = append(out, m.motionOf(k)...)
}
}
say := m.motionEvent
m.mu.Unlock()
for _, ev := range out {
say(ev)
}
}
// motionOf is the events to say now about the node and module of a model key, from their want, have and stuck key.
func (m *Model) motionOf(hk string) []MotionEvent {
w, ok := m.wants[hk]
if !ok || w.err != "" || w.value.Removed || w.value.FP == "" {
return nil
}
node, module := splitKey(hk)
h, has := m.haves[hk]
has = has && h.err == ""
body := MotionBody{Node: node, Module: module, ToFP: w.value.FP, Composed: w.value.Composed, Since: w.at, BoundSeconds: MotionBoundSeconds}
if has {
body.FromFP = h.have.FP
}
id := func(kind string) string {
return kind + "." + node + "." + module + "." + strings.TrimPrefix(w.value.Composed, "sha256:")
}
applied := has && h.have.FP == w.value.FP && (h.have.Composed == "" || h.have.Composed == w.value.Composed)
var out []MotionEvent
say := func(kind string, b MotionBody) {
if m.told[id(kind)] {
return
}
m.told[id(kind)] = true
out = append(out, MotionEvent{Type: kind, ID: id(kind), Body: b})
}
switch {
case applied && h.have.Phase == delivery.PhaseRunning && h.have.Health == delivery.HealthOK:
if m.told[id(MotionMoving)] {
b := body
b.Outcome = "running"
say(MotionSettled, b)
}
return out
case !m.told[id(MotionMoving)]:
say(MotionMoving, body)
}
stop := body
switch {
case has && h.have.Phase == delivery.PhaseHeld:
stop.Outcome, stop.Words = "held", h.have.Detail
case has && (h.have.Health == delivery.HealthFailed || h.have.Phase == delivery.PhaseFailed) && h.have.FP == w.value.FP:
stop.Outcome, stop.Words = "failed", h.have.Detail
default:
if s, ok := m.stuckWant(w.value, node, module); ok {
stop.Outcome, stop.Words = "past-bound", s
} else {
return out
}
}
say(MotionStuck, stop)
return out
}
// stuckWant is the watchman's words for the want of module on node, when it says the want stopped moving and not yet moved.
func (m *Model) stuckWant(_ delivery.Want, node, module string) (string, bool) {
wk, err := delivery.WantKey(node, module)
if err != nil {
return "", false
}
sk, err := delivery.StuckKey(delivery.AssignerWantsBucket, wk)
if err != nil {
return "", false
}
if st, ok := m.stuck[sk]; ok && st.err == "" && !st.value.Moved {
return st.value.Waited, true
}
return "", false
}
@@ -227,3 +227,122 @@ func TestADottedModuleNameIsFoundAndKept(t *testing.T) {
t.Errorf("wants = %+v", rows)
}
}
// 1906/D10 (asked by the lead for the messenger, ADR 0308): with no plans to say that a module is in motion, the view says it: `moving`
// when a module's want on a node differs from its have, `settled` when the have reaches the want running and healthy, `stuck` when it
// is held or failed or the watchman says it stopped moving; each once for a node, a module and a composition.
func motionLog(m *Model) *[]MotionEvent {
var said []MotionEvent
m.SetMotionEvent(func(e MotionEvent) { said = append(said, e) })
for _, b := range []string{delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.StuckBucket} {
m.Initial(b)
}
return &said
}
func types(said []MotionEvent) string {
var out []string
for _, e := range said {
out = append(out, e.Type)
}
return strings.Join(out, ",")
}
func TestAModuleInMotionIsSaidMovingThenSettledOnceEach(t *testing.T) {
m := New()
said := motionLog(m)
fill(m, haveFP(t, "ace", "web", fpOld, delivery.HealthOK))
if len(*said) != 0 {
t.Fatalf("said %v for a have with no want", types(*said))
}
fill(m, wantE(t, "ace", "web", delivery.Want{FP: fpNew}))
if types(*said) != "moving" {
t.Fatalf("said %q, want moving", types(*said))
}
mv := (*said)[0]
if mv.Body.Node != "ace" || mv.Body.Module != "web" || mv.Body.FromFP != fpOld || mv.Body.ToFP != fpNew || mv.Body.Composed != "sha256:"+strings.Repeat("c", 64) ||
mv.Body.Since.IsZero() || mv.Body.BoundSeconds != 1800 || mv.ID != "moving.ace.web."+strings.Repeat("c", 64) {
t.Errorf("moving = %+v", mv)
}
fill(m, haveFP(t, "ace", "web", fpOld, delivery.HealthOK), wantE(t, "ace", "web", delivery.Want{FP: fpNew}))
if types(*said) != "moving" {
t.Errorf("said %q after the same state again", types(*said))
}
fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK))
if types(*said) != "moving,settled" || (*said)[1].Body.Outcome != "running" {
t.Errorf("said %q, want settled with its outcome", types(*said))
}
fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK))
if types(*said) != "moving,settled" {
t.Errorf("settled said twice: %q", types(*said))
}
}
func TestAModuleHeldOrFailedIsSaidStuckWithItsWords(t *testing.T) {
m := New()
said := motionLog(m)
fill(m, wantE(t, "ace", "web", delivery.Want{FP: fpNew}))
b, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Phase: delivery.PhaseHeld, Health: delivery.HealthWaits, Detail: "no space left"})
fill(m, Entry{Bucket: delivery.HaveBucket, Key: "ace.web", Value: b, At: t0})
if types(*said) != "moving,stuck" || (*said)[1].Body.Outcome != "held" || (*said)[1].Body.Words != "no space left" {
t.Fatalf("said %v: %+v", types(*said), *said)
}
fill(m, Entry{Bucket: delivery.HaveBucket, Key: "ace.web", Value: b, At: t0})
if types(*said) != "moving,stuck" {
t.Errorf("stuck said twice: %q", types(*said))
}
// failed on the wanted fingerprint
m2 := New()
said2 := motionLog(m2)
fill(m2, wantE(t, "g14", "web", delivery.Want{FP: fpNew}), haveFP(t, "g14", "web", fpNew, delivery.HealthFailed))
if types(*said2) != "moving,stuck" || (*said2)[1].Body.Outcome != "failed" {
t.Errorf("failed: said %q", types(*said2))
}
}
func TestAWantTheWatchmanSaysStoppedMovingIsSaidStuckPastItsBound(t *testing.T) {
m := New()
said := motionLog(m)
fill(m, wantE(t, "ace", "web", delivery.Want{FP: fpNew}), haveFP(t, "ace", "web", fpOld, delivery.HealthOK))
sk, _ := delivery.StuckKey(delivery.AssignerWantsBucket, "want.ace.web")
v, err := delivery.EncodeStuck(delivery.Stuck{Bucket: delivery.AssignerWantsBucket, Key: "want.ace.web", Waited: "the node agent of ace to run web", Since: t0})
if err != nil {
t.Fatal(err)
}
fill(m, Entry{Bucket: delivery.StuckBucket, Key: sk, Value: v, At: t0})
if types(*said) != "moving,stuck" || (*said)[1].Body.Outcome != "past-bound" {
t.Errorf("said %q: %+v", types(*said), *said)
}
}
func TestANodeThatSettlesWithoutBeingSeenMovingSaysNothing(t *testing.T) {
m := New()
said := motionLog(m)
fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK), wantE(t, "ace", "web", delivery.Want{FP: fpNew}))
if len(*said) != 0 {
t.Errorf("said %v for a module that never moved", types(*said))
}
}
// Nothing is said of a module in motion before the buckets have been read: a want read ahead of its have is not a node behind. What is
// in motion when they have been read is said then, and a module that settles after is said settled.
func TestMotionIsSaidOnlyOnceTheBucketsHaveBeenRead(t *testing.T) {
m := New()
var said []MotionEvent
m.SetMotionEvent(func(e MotionEvent) { said = append(said, e) })
fill(m, wantE(t, "ace", "web", delivery.Want{FP: fpNew}), wantE(t, "ace", "mail", delivery.Want{FP: fpNew}))
m.Initial(delivery.AssignerWantsBucket)
fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK), haveFP(t, "ace", "mail", fpOld, delivery.HealthOK))
m.Initial(delivery.HaveBucket)
if len(said) != 0 {
t.Fatalf("said %v before the stuck keys were read", types(said))
}
m.Initial(delivery.StuckBucket)
if len(said) != 1 || said[0].Type != MotionMoving || said[0].Body.Module != "mail" {
t.Fatalf("said %v: %+v, want only mail moving (web is already there)", types(said), said)
}
fill(m, haveFP(t, "ace", "mail", fpNew, delivery.HealthOK))
if types(said) != "moving,settled" {
t.Errorf("said %q", types(said))
}
}
+2 -2
View File
@@ -96,8 +96,8 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
}
}
// The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves.
if !reflect.DeepEqual(m.Emits, []string{"delivered"}) {
t.Errorf("emits = %v, want exactly [delivered]", m.Emits)
if !reflect.DeepEqual(m.Emits, []string{"delivered", "moving", "settled", "stuck"}) {
t.Errorf("emits = %v, want exactly delivered, moving, settled and stuck (1906/D9, D10)", m.Emits)
}
// The view uses no build seat: it reads their workers by the broker's named grant and publishes no ask (ADR 0525 §2).
if len(m.Uses) != 0 {
+4 -1
View File
@@ -16,7 +16,10 @@
"node_haves"
],
"emits": [
"delivered"
"delivered",
"moving",
"settled",
"stuck"
],
"state": [
{ "name": "answers", "history": 5 }