novox/mesh-controller
18
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Keep a condition's looks in the store, so a look that changes nothing writes no bucket (issue 420)
mesh/merge-gate pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus step; every machine composes with the change as it…
Details
mesh/repo-check pass: its merge-check.sh passed
Details
mesh/delivery delivered
Details

Browse Source 
Every look rewrote the condition's value in the conditions bucket, because the
value carried the count of looks, the last look and the newest evidence: 131 647
writes in four days for twelve open conditions, against ADR 0289's rule that a
bucket is written only when what it describes changed. The looks now live in the
controller's store (migration 0095, condition_looks), the bucket holds what a
condition says and is written only when that changes, and the conditions verb and
the condition events put the looks back.
This commit is contained in:
jochen
2026-10-12 12:11:13 +02:00
parent 22a2f73f7b
commit dcdafd37f0
18 changed files with 803 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/conditions/store.go
+94 -16
View File
@@ -1,6 +1,7 @@
package conditions
import (
"bytes"
"context"
"encoding/json"
"errors"
@@ -53,6 +54,7 @@ const Seat = "mesh-controller"
// Keeper raises, observes, silences and clears conditions, and says each transition.
type Keeper struct {
store Backend
looks LookStore
history History
teller Teller
now func() time.Time
@@ -91,7 +93,10 @@ type clearing struct {
// Options are what a Keeper is made with.
type Options struct {
Store Backend
Store Backend
// Looks is where the open conditions' looks are kept (looks.go): the controller's store. Nil takes the
// Store's own when it keeps them (the in-memory store does), and otherwise keeps them in this process.
Looks LookStore
History History
// Teller says the transitions; nil says nothing (a test, or a command run with no bus to say on).
Teller Teller
@@ -111,7 +116,7 @@ var TellFor = 10 * time.Minute
// NewKeeper is a keeper over a store. It reads what cleared lately from the history, so a condition
// that cleared just before this controller started and is raised again now is a reopening.
func NewKeeper(ctx context.Context, o Options) *Keeper {
k := &Keeper{store: o.Store, history: o.History, teller: o.Teller, now: o.Now, say: o.Say, changed: o.Changed,
k := &Keeper{store: o.Store, looks: o.Looks, history: o.History, teller: o.Teller, now: o.Now, say: o.Say, changed: o.Changed,
epoch: o.Epoch,
cleared: map[string]clearing{}, reworded: map[string]bool{}, out: make(chan Event, 1024), drained: make(chan struct{})}
if k.now == nil {
@@ -120,6 +125,13 @@ func NewKeeper(ctx context.Context, o Options) *Keeper {
if k.say == nil {
k.say = func(string, ...any) {}
}
if k.looks == nil {
if own, ok := o.Store.(LookStore); ok {
k.looks = own
} else {
k.looks = NewMemoryLooks()
}
}
if k.history != nil {
if recent, err := k.history.Since(ctx, k.now().Add(-ReopenWithin)); err == nil {
for _, e := range recent {
@@ -218,7 +230,7 @@ func (k *Keeper) Observe(ctx context.Context, o Observation) (Condition, error)
if err := k.stamp(&c); err != nil {
return Condition{}, err
}
body, err := json.Marshal(c)
body, err := BucketValue(c)
if err != nil {
return Condition{}, err
}
@@ -227,6 +239,7 @@ func (k *Keeper) Observe(ctx context.Context, o Observation) (Condition, error)
} else if err != nil {
return Condition{}, fmt.Errorf("raising the condition %s: %w", key, err)
}
k.keepLooks(ctx, &c)
k.mu.Lock()
delete(k.cleared, key)
k.mu.Unlock()
@@ -237,6 +250,7 @@ func (k *Keeper) Observe(ctx context.Context, o Observation) (Condition, error)
if err := json.Unmarshal(entry.Value, &c); err != nil {
return Condition{}, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
}
looked := k.attach(ctx, &c)
var changes []Event
// A healer's budget spent made it urgent; the watchdog seeing it again does not undo that.
if o.Severity != c.Severity && !(c.Escalated() && o.Severity != Urgent) {
@@ -278,14 +292,22 @@ func (k *Keeper) Observe(ctx context.Context, o Observation) (Condition, error)
if err := k.stamp(&c); err != nil {
return Condition{}, err
}
body, err := json.Marshal(c)
body, err := BucketValue(c)
if err != nil {
return Condition{}, err
}
if err := k.store.Update(ctx, key, body, entry.Revision); errors.Is(err, ErrMoved) {
continue
} else if err != nil {
return Condition{}, fmt.Errorf("observing the condition %s: %w", key, err)
// Written only when what the condition says changed (novox/hq ADR 0289 §1, issue 420): a look
// that saw it as it was changes only its looks, which the controller's store keeps.
if !bytes.Equal(body, entry.Value) {
if err := k.store.Update(ctx, key, body, entry.Revision); errors.Is(err, ErrMoved) {
continue
} else if err != nil {
return Condition{}, fmt.Errorf("observing the condition %s: %w", key, err)
}
}
// Looks read as unreadable are not written over: a count of one would replace the count kept.
if fresh && looked {
k.keepLooks(ctx, &c)
}
for _, e := range changes {
e.At, e.Condition = now, c
@@ -318,6 +340,7 @@ func (k *Keeper) ClearSaying(ctx context.Context, key, why, resolved string) (bo
// Unreadable is not resolved: kept, and said, rather than removed unread.
return false, fmt.Errorf("the condition %s on the bus cannot be read, so it is not cleared: %w", key, err)
}
k.attach(ctx, &c)
if err := k.stamp(&c); err != nil {
return false, err
}
@@ -329,6 +352,10 @@ func (k *Keeper) ClearSaying(ctx context.Context, key, why, resolved string) (bo
} else if err != nil {
return false, fmt.Errorf("clearing the condition %s: %w", key, err)
}
if err := k.looks.ForgetLooks(ctx, key); err != nil {
k.say("the observations of the condition %s, which cleared, could not be forgotten in the controller's "+
"store; a raising of it counts from one all the same: %v", key, err)
}
now := k.now().UTC()
k.mu.Lock()
k.cleared[key] = clearing{at: now, count: c.Count, began: c.Began(), gaps: c.Gaps, silenced: c.Silenced, tried: c.Tried}
@@ -402,12 +429,13 @@ func (k *Keeper) amend(ctx context.Context, key string, change func(*Condition,
if err := json.Unmarshal(entry.Value, &c); err != nil {
return Condition{}, false, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
}
k.attach(ctx, &c)
now := k.now().UTC()
changes := change(&c, now)
if err := k.stamp(&c); err != nil {
return Condition{}, false, err
}
body, err := json.Marshal(c)
body, err := BucketValue(c)
if err != nil {
return Condition{}, false, err
}
@@ -479,12 +507,13 @@ func (k *Keeper) Silence(ctx context.Context, key string, d time.Duration, by, w
if err := json.Unmarshal(entry.Value, &c); err != nil {
return Condition{}, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
}
k.attach(ctx, &c)
now := k.now().UTC()
c.Silenced = &Silence{Until: now.Add(d), By: by, Why: strings.TrimSpace(why), Since: now}
if err := k.stamp(&c); err != nil {
return Condition{}, err
}
body, err := json.Marshal(c)
body, err := BucketValue(c)
if err != nil {
return Condition{}, err
}
@@ -526,12 +555,13 @@ func (k *Keeper) EndSilences(ctx context.Context) error {
if held.Silenced == nil || now.Before(held.Silenced.Until) {
break
}
k.attach(ctx, &held)
was := held.Silenced.Why
held.Silenced = nil
if err := k.stamp(&held); err != nil {
return err
}
body, err := json.Marshal(held)
body, err := BucketValue(held)
if err != nil {
return err
}
@@ -549,12 +579,12 @@ func (k *Keeper) EndSilences(ctx context.Context) error {
// Open is every open condition, urgent first and then oldest first.
func (k *Keeper) Open(ctx context.Context) ([]Condition, error) {
return Read(ctx, k.store)
return ReadWith(ctx, k.store, k.looks)
}
// Get is one open condition.
func (k *Keeper) Get(ctx context.Context, key string) (Condition, bool, error) {
return ReadOne(ctx, k.store, key)
return ReadOneWith(ctx, k.store, k.looks, key)
}
// HistorySince is every transition from a moment, oldest first.
@@ -565,9 +595,16 @@ func (k *Keeper) HistorySince(ctx context.Context, since time.Time) ([]Event, er
return k.history.Since(ctx, since)
}
// Read is every open condition in a store, in the order status says them. A value that cannot be
// read is an error naming its key, never a condition left out (ADR 0227 rule 4).
// Read is every open condition in a store, in the order status says them, with its looks when the
// store keeps them. A value that cannot be read is an error naming its key, never a condition left out
// (ADR 0227 rule 4).
func Read(ctx context.Context, store Backend) ([]Condition, error) {
looks, _ := store.(LookStore)
return ReadWith(ctx, store, looks)
}
// ReadWith is Read with each condition's looks from where they are kept (looks.go); nil reads none.
func ReadWith(ctx context.Context, store Backend, looks LookStore) ([]Condition, error) {
all, err := store.All(ctx)
if err != nil {
return nil, fmt.Errorf("the open conditions cannot be read: %w", err)
@@ -580,12 +617,19 @@ func Read(ctx context.Context, store Backend) ([]Condition, error) {
}
out = append(out, c)
}
attachLooks(ctx, looks, out)
Order(out)
return out, nil
}
// ReadOne is one open condition from a store.
// ReadOne is one open condition from a store, with its looks when the store keeps them.
func ReadOne(ctx context.Context, store Backend, key string) (Condition, bool, error) {
looks, _ := store.(LookStore)
return ReadOneWith(ctx, store, looks, key)
}
// ReadOneWith is ReadOne with its looks from where they are kept; nil reads none.
func ReadOneWith(ctx context.Context, store Backend, looks LookStore, key string) (Condition, bool, error) {
e, found, err := store.Get(ctx, key)
if err != nil || !found {
return Condition{}, found, err
@@ -594,9 +638,43 @@ func ReadOne(ctx context.Context, store Backend, key string) (Condition, bool, e
if err := json.Unmarshal(e.Value, &c); err != nil {
return Condition{}, false, fmt.Errorf("the condition %s cannot be read: %w", key, err)
}
if looks != nil {
l, kept, err := looks.Looked(ctx, key)
if err != nil {
c.LooksUnread = "its observations could not be read from the controller's store: " + err.Error()
} else {
attachOne(&c, l, kept)
}
}
return c, true, nil
}
// attach puts a condition's kept looks on it, as read from the bucket. False when they could not be
// read, which is said: the condition then carries why, and its looks are not written over.
func (k *Keeper) attach(ctx context.Context, c *Condition) bool {
l, kept, err := k.looks.Looked(ctx, c.Key)
if err != nil {
c.LooksUnread = "its observations could not be read from the controller's store: " + err.Error()
k.say("the observations of the condition %s could not be read from the controller's store, so this look "+
"is not counted: %v", c.Key, err)
return false
}
attachOne(c, l, kept)
return true
}
// keepLooks writes a condition's looks, saying a write that failed: the condition is raised or kept in the
// bucket all the same, and this look is not counted.
func (k *Keeper) keepLooks(ctx context.Context, c *Condition) {
if err := k.looks.KeepLooks(ctx, c.Key, c.looks()); err == nil {
c.LooksUnread = ""
} else {
c.LooksUnread = "its observations could not be kept in the controller's store: " + err.Error()
k.say("the observations of the condition %s could not be kept in the controller's store, so this look "+
"is not counted: %v", c.Key, err)
}
}
// tell queues a transition to be kept and said. Never blocks the caller for long: a queue that is
// full is a bus away for a long time, and the transition is said lost rather than holding a watchdog.
func (k *Keeper) tell(e Event) {