Merge pull request 'Keep a condition's looks in the store, so a look that changes nothing writes no bucket (issue 420)' (#249) from fix/420-conditions-bucket-on-transition into main
This commit was merged in pull request #249.
This commit is contained in:
@@ -0,0 +1,76 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// The looks of the open conditions — how often each was observed, when last, its newest evidence — are
|
||||
// kept in the controller's store, not the conditions bucket (novox/hq issue 420, ADR 0289 §1): the bucket
|
||||
// is written when what a condition says changes, never for a look that saw it as it was.
|
||||
|
||||
// storeLooks keeps the conditions' looks in the inventory.
|
||||
type storeLooks struct{ inv *inventory.Inventory }
|
||||
|
||||
func (s storeLooks) KeepLooks(ctx context.Context, key string, l conditions.Looks) error {
|
||||
evidence, err := json.Marshal(l.Evidence)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.inv.KeepConditionLooks(ctx, inventory.ConditionLooks{Key: key, Observations: l.Observations,
|
||||
LastObserved: l.LastObserved, Evidence: evidence})
|
||||
}
|
||||
|
||||
func (s storeLooks) Looked(ctx context.Context, key string) (conditions.Looks, bool, error) {
|
||||
row, kept, err := s.inv.ConditionLooksOf(ctx, key)
|
||||
if err != nil || !kept {
|
||||
return conditions.Looks{}, kept, err
|
||||
}
|
||||
l, err := looksOf(row)
|
||||
return l, err == nil, err
|
||||
}
|
||||
|
||||
func (s storeLooks) AllLooks(ctx context.Context) (map[string]conditions.Looks, error) {
|
||||
rows, err := s.inv.AllConditionLooks(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make(map[string]conditions.Looks, len(rows))
|
||||
for key, row := range rows {
|
||||
l, err := looksOf(row)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[key] = l
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s storeLooks) ForgetLooks(ctx context.Context, key string) error {
|
||||
return s.inv.ForgetConditionLooks(ctx, key)
|
||||
}
|
||||
|
||||
func looksOf(row inventory.ConditionLooks) (conditions.Looks, error) {
|
||||
l := conditions.Looks{Observations: row.Observations, LastObserved: row.LastObserved.UTC()}
|
||||
if len(row.Evidence) > 0 {
|
||||
if err := json.Unmarshal(row.Evidence, &l.Evidence); err != nil {
|
||||
return conditions.Looks{}, fmt.Errorf("the evidence kept for the condition %s cannot be read: %w", row.Key, err)
|
||||
}
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
// looksHere is where a process that is not the serving controller reads the conditions' looks: the store,
|
||||
// opened for it, closed by the function returned. A store that cannot be opened is said on every condition
|
||||
// read through it, never read as none.
|
||||
func looksHere(ctx context.Context) (conditions.LookStore, func()) {
|
||||
inv, err := openInventory(ctx)
|
||||
if err != nil {
|
||||
return conditions.NoLooks{Err: err}, func() {}
|
||||
}
|
||||
return storeLooks{inv}, inv.Close
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// **Looks at an unchanged condition write the bucket once, and `conditions` still says how often and when
|
||||
// it was seen** (novox/hq issue 420): the looks are in the controller's store, and the verb, run as a
|
||||
// process of its own, reads them from there. On 2026-10-10 one open condition had rewritten the bucket 7 003
|
||||
// times in two days.
|
||||
func TestTheConditionsVerbReadsTheLooksTheBucketNoLongerHolds(t *testing.T) {
|
||||
inv := inventory.ForTest(t)
|
||||
bus := testbus.Start(t)
|
||||
serving, err := broker.Dial(bus.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer serving.Close()
|
||||
if err := serving.EnsureControllerBuckets(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := t.Context()
|
||||
store, history, err := conditions.OnTheBus(ctx, serving.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
k := conditions.NewKeeper(ctx, conditions.Options{Store: store, History: history, Looks: storeLooks{inv}})
|
||||
defer k.Close(context.Background())
|
||||
look := conditions.Observation{Scope: conditions.ScopeMachine, ID: "anchor", Kind: "silent", Machine: "anchor",
|
||||
Severity: conditions.Warning, Summary: "anchor has not been heard from", Source: "S1"}
|
||||
for range 4 {
|
||||
if _, err := k.Observe(ctx, look); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
api, err := jetstream.New(serving.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stream, err := api.Stream(ctx, "KV_"+broker.ConditionsBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := stream.Info(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.State.LastSeq != 1 {
|
||||
t.Fatalf("four looks at one unchanged condition wrote the bucket %d times, want 1", info.State.LastSeq)
|
||||
}
|
||||
|
||||
// The verb as a process other than the serving controller: it reads the bucket and the store.
|
||||
keptBefore := conditionsFrom
|
||||
conditionsFrom = nil
|
||||
defer func() { conditionsFrom = keptBefore }()
|
||||
t.Setenv(broker.NATSVar, bus.ClientURL())
|
||||
t.Setenv(broker.CertificateVar, "")
|
||||
var shown bytes.Buffer
|
||||
if err := showCondition(ctx, []string{"machine.anchor.silent", "--json"}, &shown); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var c conditions.Condition
|
||||
if err := json.Unmarshal(shown.Bytes(), &c); err != nil {
|
||||
t.Fatalf("%v: %s", err, shown.String())
|
||||
}
|
||||
if c.Observations != 4 || len(c.Evidence) != 4 || c.LastObserved.IsZero() || c.LooksUnread != "" ||
|
||||
time.Since(c.LastObserved) > time.Minute {
|
||||
t.Fatalf("conditions show reads %d observation(s), %d evidence line(s), last %s (%q)", c.Observations,
|
||||
len(c.Evidence), c.LastObserved, c.LooksUnread)
|
||||
}
|
||||
listed, err := openConditions(ctx)
|
||||
if err != nil || len(listed) != 1 || listed[0].Observations != 4 {
|
||||
t.Fatalf("conditions lists %+v: %v", listed, err)
|
||||
}
|
||||
}
|
||||
@@ -24,16 +24,18 @@ import (
|
||||
// **`status` used to be the only place the mesh said it was wrong, and only to whoever asked.** Every
|
||||
// core failure of research 031 was found by a person looking. A condition is the mesh saying it: raised
|
||||
// by a watchdog when a signal is late (signals.go), by a probe when an invariant does not hold
|
||||
// (doctor.go), or by an event a provider sends (standing.go); kept on the bus with since-when and
|
||||
// evidence; said on the bus as it changes, for the operator's channel to carry; and cleared when an
|
||||
// (doctor.go), or by an event a provider sends (standing.go); kept on the bus with since-when, and its
|
||||
// looks — how often, when last, the newest evidence — in the controller's store (condition_looks.go,
|
||||
// novox/hq issue 420); said on the bus as it changes, for the operator's channel to carry; and cleared when an
|
||||
// observation says it is resolved. Nobody resolves one by hand. A person who knows silences it, for a
|
||||
// while, with a reason, and that is recorded as a hand act.
|
||||
|
||||
// conditionsFrom is the serving controller's keeper; nil in any other process, which opens its own.
|
||||
var conditionsFrom *conditions.Keeper
|
||||
|
||||
// keeperOn is a keeper over the store on a connection, saying its transitions on that connection.
|
||||
func keeperOn(ctx context.Context, conn *nats.Conn) (*conditions.Keeper, error) {
|
||||
// keeperOn is a keeper over the store on a connection, saying its transitions on that connection, with
|
||||
// the conditions' looks kept in looks (condition_looks.go).
|
||||
func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore) (*conditions.Keeper, error) {
|
||||
store, history, err := conditions.OnTheBus(ctx, conn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -42,7 +44,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn) (*conditions.Keeper, error)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return conditions.NewKeeper(ctx, conditions.Options{Store: store, History: history,
|
||||
return conditions.NewKeeper(ctx, conditions.Options{Store: store, History: history, Looks: looks,
|
||||
Teller: link.OverNATS{Conn: conn, JS: js},
|
||||
Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
|
||||
// What status leads with changed: composed again soon (a nudge outside the serving controller
|
||||
@@ -58,8 +60,10 @@ func withKeeper(ctx context.Context, f func(*conditions.Keeper) error) error {
|
||||
if conditionsFrom != nil {
|
||||
return f(conditionsFrom)
|
||||
}
|
||||
looks, closeLooks := looksHere(ctx)
|
||||
defer closeLooks()
|
||||
return onTheBus(func(conn *nats.Conn) error {
|
||||
k, err := keeperOn(ctx, conn)
|
||||
k, err := keeperOn(ctx, conn, looks)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -82,6 +86,8 @@ func openConditions(ctx context.Context) ([]conditions.Condition, error) {
|
||||
return nil, fmt.Errorf("this process has no bus to read the conditions from: %w", err)
|
||||
}
|
||||
var out []conditions.Condition
|
||||
looks, closeLooks := looksHere(ctx)
|
||||
defer closeLooks()
|
||||
err := onTheBus(func(conn *nats.Conn) error {
|
||||
store, _, err := conditions.OnTheBus(ctx, conn)
|
||||
if err != nil {
|
||||
@@ -89,7 +95,7 @@ func openConditions(ctx context.Context) ([]conditions.Condition, error) {
|
||||
}
|
||||
reading, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
out, err = conditions.Read(reading, store)
|
||||
out, err = conditions.ReadWith(reading, store, looks)
|
||||
return err
|
||||
})
|
||||
return out, err
|
||||
@@ -252,12 +258,14 @@ func showCondition(ctx context.Context, args []string, w io.Writer) error {
|
||||
if conditionsFrom != nil {
|
||||
c, found, err = conditionsFrom.Get(ctx, key)
|
||||
} else {
|
||||
looks, closeLooks := looksHere(ctx)
|
||||
defer closeLooks()
|
||||
err = onTheBus(func(conn *nats.Conn) error {
|
||||
store, _, err := conditions.OnTheBus(ctx, conn)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c, found, err = conditions.ReadOne(ctx, store, key)
|
||||
c, found, err = conditions.ReadOneWith(ctx, store, looks, key)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
@@ -651,7 +651,7 @@ func localDoctor(ctx context.Context) (*doctor, func(), error) {
|
||||
open.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
k, err := keeperOn(ctx, js.Conn())
|
||||
k, err := keeperOn(ctx, js.Conn(), storeLooks{open.inventory})
|
||||
if err != nil {
|
||||
js.Close()
|
||||
open.Close()
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
@@ -30,7 +31,7 @@ func TestReplay327(t *testing.T) {
|
||||
if err := serving.EnsureControllerBuckets(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keeper, err := keeperOn(context.Background(), serving.Conn())
|
||||
keeper, err := keeperOn(context.Background(), serving.Conn(), conditions.NewMemoryLooks())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -704,7 +704,7 @@ func later(a, b time.Time) time.Time {
|
||||
// self-check, for the serving controller; the returned function stops them. Nil when the store could
|
||||
// not be opened, which is said.
|
||||
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() {
|
||||
keeper, err := keeperOn(ctx, bus.Conn)
|
||||
keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory})
|
||||
if err != nil {
|
||||
fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+
|
||||
"status says the conditions cannot be read: %v\n", err)
|
||||
|
||||
@@ -108,7 +108,8 @@ var WritersTable = []WriterRow{
|
||||
{State: "plans and their tiers", Writer: "controller (lease holder), compare-and-set on the plan's revision",
|
||||
KeptIn: "the controller's store", Others: "read through plans"},
|
||||
{State: "conditions", Writer: "controller", KeptIn: "key-value " + ConditionsBucket + " (and its history, " +
|
||||
ConditionHistoryBucket + ")", Others: "raise or clear only through observations the controller reads",
|
||||
ConditionHistoryBucket + "); how often each was observed, when last and its newest evidence in the " +
|
||||
"controller's store, so a look that changes nothing writes no bucket", Others: "raise or clear only through observations the controller reads",
|
||||
Subjects: append(kvOf(ConditionsBucket), kvOf(ConditionHistoryBucket)...), Writes: isController},
|
||||
{State: "calls and their outcomes", Writer: "controller", KeptIn: "key-value " + CallsBucket,
|
||||
Others: "read by id", Subjects: kvOf(CallsBucket), Writes: isController},
|
||||
|
||||
@@ -44,8 +44,8 @@ func busStoreForTest(t *testing.T) (*broker.JetStream, Backend, History) {
|
||||
func TestNatsTheStoreKeepsConditionsByCompareAndSet(t *testing.T) {
|
||||
_, store, history := busStoreForTest(t)
|
||||
ctx := t.Context()
|
||||
told := &Told{}
|
||||
k := NewKeeper(ctx, Options{Store: store, History: history, Teller: told})
|
||||
told, looks := &Told{}, NewMemoryLooks()
|
||||
k := NewKeeper(ctx, Options{Store: store, History: history, Teller: told, Looks: looks})
|
||||
if _, err := k.Observe(ctx, silent("ace")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -54,7 +54,7 @@ func TestNatsTheStoreKeepsConditionsByCompareAndSet(t *testing.T) {
|
||||
}
|
||||
k.Close(context.Background())
|
||||
|
||||
again := NewKeeper(ctx, Options{Store: store, History: history})
|
||||
again := NewKeeper(ctx, Options{Store: store, History: history, Looks: looks})
|
||||
defer again.Close(context.Background())
|
||||
open, err := again.Open(ctx)
|
||||
if err != nil {
|
||||
@@ -67,13 +67,13 @@ func TestNatsTheStoreKeepsConditionsByCompareAndSet(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Update(ctx, "machine.ace.silent", []byte(`{}`), e.Revision-1); err != ErrMoved {
|
||||
if err := store.Update(ctx, "machine.ace.silent", []byte(`{}`), e.Revision+7); err != ErrMoved {
|
||||
t.Fatalf("a write at a stale revision answered %v", err)
|
||||
}
|
||||
if err := store.Create(ctx, "machine.ace.silent", []byte(`{}`)); err != ErrMoved {
|
||||
t.Fatalf("creating an open condition answered %v", err)
|
||||
}
|
||||
if err := store.Delete(ctx, "machine.ace.silent", e.Revision-1); err != ErrMoved {
|
||||
if err := store.Delete(ctx, "machine.ace.silent", e.Revision+7); err != ErrMoved {
|
||||
t.Fatalf("a delete at a stale revision answered %v", err)
|
||||
}
|
||||
if cleared, err := again.Clear(ctx, "machine.ace.silent", "heard"); err != nil || !cleared {
|
||||
|
||||
@@ -138,7 +138,9 @@ type Condition struct {
|
||||
// opens "Nothing for you to do."). Actions are the answers a notification offers (plain.go).
|
||||
Needs string `json:"needs"`
|
||||
Actions []Action `json:"actions"`
|
||||
// Evidence is the newest observations, at most KeptEvidence, newest first.
|
||||
// Evidence is the newest observations, at most KeptEvidence, newest first. It, LastObserved and
|
||||
// Observations are the condition's looks (looks.go): kept in the controller's store, never in the
|
||||
// conditions bucket, and put back on every condition the keeper reads and every event it says.
|
||||
Evidence []Evidence `json:"evidence"`
|
||||
// Source is the signals-table row, probe or event that raised it: `S1`, `D3`, `provisioner.failing`.
|
||||
Source string `json:"source"`
|
||||
@@ -167,6 +169,9 @@ type Condition struct {
|
||||
// Epoch is the controller lease epoch that last wrote it (to-be 45 §6). Zero where it was written
|
||||
// by a controller serving without the lease, or before the lease existed.
|
||||
Epoch uint64 `json:"epoch"`
|
||||
// LooksUnread says why Observations, LastObserved and Evidence are not known: they are kept in the
|
||||
// controller's store, not the bucket (looks.go), and could not be read from it. Empty when they were.
|
||||
LooksUnread string `json:"looks-unread,omitempty"`
|
||||
}
|
||||
|
||||
// SilencedAt says whether a person's silence is in force at a moment.
|
||||
|
||||
@@ -0,0 +1,163 @@
|
||||
package conditions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Looks are what an open condition's observations say between its transitions: how many there were,
|
||||
// when the newest was, and the newest evidence (to-be 45 §2). They change on every look, so **they are
|
||||
// not kept in the conditions bucket**, which is written only when what a condition says changes
|
||||
// (novox/hq ADR 0289 §1 and §4, issue 420): a watcher of the bucket heard thousands of writes a day for a
|
||||
// dozen conditions that did not change. The controller's store keeps them (LookStore), and the keeper
|
||||
// puts them back on every condition it reads and every event it says, so `conditions` and the events
|
||||
// still carry them.
|
||||
type Looks struct {
|
||||
Observations int `json:"observations"`
|
||||
LastObserved time.Time `json:"last-observed"`
|
||||
Evidence []Evidence `json:"evidence"`
|
||||
}
|
||||
|
||||
// LookStore is where the looks of the open conditions are kept, one row per key.
|
||||
type LookStore interface {
|
||||
// KeepLooks writes a condition's looks, replacing what it held.
|
||||
KeepLooks(ctx context.Context, key string, l Looks) error
|
||||
// Looked is one condition's looks; false when none are kept.
|
||||
Looked(ctx context.Context, key string) (Looks, bool, error)
|
||||
// AllLooks is every condition's looks. An error is an error, never "none kept" (ADR 0227 rule 4).
|
||||
AllLooks(ctx context.Context) (map[string]Looks, error)
|
||||
// ForgetLooks removes a condition's looks: it cleared.
|
||||
ForgetLooks(ctx context.Context, key string) error
|
||||
}
|
||||
|
||||
// lookFields are a condition's fields that are its looks, or say they could not be read: left out of
|
||||
// the bucket's value.
|
||||
var lookFields = []string{"observations", "last-observed", "evidence", "looks-unread"}
|
||||
|
||||
// looks is the condition's looks.
|
||||
func (c Condition) looks() Looks {
|
||||
return Looks{Observations: c.Observations, LastObserved: c.LastObserved, Evidence: c.Evidence}
|
||||
}
|
||||
|
||||
// withLooks puts looks on the condition.
|
||||
func (c *Condition) withLooks(l Looks) {
|
||||
c.Observations, c.LastObserved, c.Evidence = l.Observations, l.LastObserved, l.Evidence
|
||||
}
|
||||
|
||||
// BucketValue is a condition as the conditions bucket keeps it: everything but its looks. Its keys are in
|
||||
// one order, so the same condition is the same bytes and an unchanged one is not written again.
|
||||
func BucketValue(c Condition) ([]byte, error) {
|
||||
whole, err := json.Marshal(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
if err := json.Unmarshal(whole, &fields); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, f := range lookFields {
|
||||
delete(fields, f)
|
||||
}
|
||||
return json.Marshal(fields)
|
||||
}
|
||||
|
||||
// attachLooks puts each condition's kept looks on it. A condition the store keeps no looks for keeps
|
||||
// what its value says (one written before the looks moved out of the bucket says them); one that says
|
||||
// none is said to have none kept. Looks that cannot be read are said on each condition, never left
|
||||
// as a count of nothing.
|
||||
func attachLooks(ctx context.Context, looks LookStore, list []Condition) {
|
||||
if looks == nil {
|
||||
return
|
||||
}
|
||||
all, err := looks.AllLooks(ctx)
|
||||
for i := range list {
|
||||
if err != nil {
|
||||
list[i].LooksUnread = "its observations could not be read from the controller's store: " + err.Error()
|
||||
continue
|
||||
}
|
||||
l, found := all[list[i].Key]
|
||||
attachOne(&list[i], l, found)
|
||||
}
|
||||
}
|
||||
|
||||
func attachOne(c *Condition, l Looks, found bool) {
|
||||
switch {
|
||||
case found:
|
||||
c.withLooks(l)
|
||||
case c.LastObserved.IsZero():
|
||||
c.LooksUnread = "the controller's store keeps no observations for it"
|
||||
}
|
||||
}
|
||||
|
||||
// MemoryLooks keeps looks in this process: for tests, and for a keeper given no store of looks.
|
||||
type MemoryLooks struct {
|
||||
mu sync.Mutex
|
||||
looks map[string]Looks
|
||||
// Kept counts the writes, so a test can say a look was kept.
|
||||
kept int
|
||||
}
|
||||
|
||||
// NewMemoryLooks is an empty store of looks.
|
||||
func NewMemoryLooks() *MemoryLooks { return &MemoryLooks{looks: map[string]Looks{}} }
|
||||
|
||||
func (m *MemoryLooks) KeepLooks(_ context.Context, key string, l Looks) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.looks == nil {
|
||||
m.looks = map[string]Looks{}
|
||||
}
|
||||
l.Evidence = append([]Evidence(nil), l.Evidence...)
|
||||
m.looks[key] = l
|
||||
m.kept++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MemoryLooks) Looked(_ context.Context, key string) (Looks, bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
l, ok := m.looks[key]
|
||||
l.Evidence = append([]Evidence(nil), l.Evidence...)
|
||||
return l, ok, nil
|
||||
}
|
||||
|
||||
func (m *MemoryLooks) AllLooks(context.Context) (map[string]Looks, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
out := make(map[string]Looks, len(m.looks))
|
||||
for k, l := range m.looks {
|
||||
l.Evidence = append([]Evidence(nil), l.Evidence...)
|
||||
out[k] = l
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (m *MemoryLooks) ForgetLooks(_ context.Context, key string) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
delete(m.looks, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// LooksKept is how many times looks were written.
|
||||
func (m *MemoryLooks) LooksKept() int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.kept
|
||||
}
|
||||
|
||||
// NoLooks is a store of looks that cannot be reached, saying why on every read and write: a command
|
||||
// whose process could not open the controller's store still reads and writes the bucket, and says its
|
||||
// observations unread rather than none.
|
||||
type NoLooks struct{ Err error }
|
||||
|
||||
func (n NoLooks) err() error { return fmt.Errorf("the controller's store is not open here: %w", n.Err) }
|
||||
|
||||
func (n NoLooks) KeepLooks(context.Context, string, Looks) error { return n.err() }
|
||||
func (n NoLooks) Looked(context.Context, string) (Looks, bool, error) {
|
||||
return Looks{}, false, n.err()
|
||||
}
|
||||
func (n NoLooks) AllLooks(context.Context) (map[string]Looks, error) { return nil, n.err() }
|
||||
func (n NoLooks) ForgetLooks(context.Context, string) error { return n.err() }
|
||||
@@ -0,0 +1,161 @@
|
||||
package conditions
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// **A look that sees a condition as it was writes nothing to the bucket** (novox/hq issue 420, ADR 0289
|
||||
// §1 and §4): on 2026-10-10 the bucket took 131 647 writes in four days for twelve open conditions,
|
||||
// because every look rewrote the count, the last look and the evidence. The bucket is written when the
|
||||
// condition is raised and when what it says changes; the looks are kept beside it, and `conditions` and
|
||||
// the events still carry them.
|
||||
func TestALookThatChangesNothingWritesNothingToTheBucket(t *testing.T) {
|
||||
k, store, told, c := keeper(t)
|
||||
ctx := t.Context()
|
||||
for range 5 {
|
||||
if _, err := k.Observe(ctx, silent("ace")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.pass(30 * time.Second)
|
||||
}
|
||||
if store.Writes() != 1 {
|
||||
t.Fatalf("five looks at one unchanged condition wrote the bucket %d times, want 1 (its raising)", store.Writes())
|
||||
}
|
||||
got, open, err := k.Get(ctx, "machine.ace.silent")
|
||||
if err != nil || !open {
|
||||
t.Fatalf("open %v: %v", open, err)
|
||||
}
|
||||
if got.Observations != 5 || len(got.Evidence) != 5 || !got.LastObserved.Equal(c.now().Add(-30*time.Second)) ||
|
||||
got.LooksUnread != "" {
|
||||
t.Fatalf("the verb reads %d observation(s), %d evidence line(s), last %s (%q)", got.Observations,
|
||||
len(got.Evidence), got.LastObserved, got.LooksUnread)
|
||||
}
|
||||
list, err := k.Open(ctx)
|
||||
if err != nil || len(list) != 1 || list[0].Observations != 5 {
|
||||
t.Fatalf("the list reads %+v: %v", list, err)
|
||||
}
|
||||
|
||||
// The bucket's value holds what the condition says, and none of its looks.
|
||||
e, _, _ := store.Get(ctx, "machine.ace.silent")
|
||||
var fields map[string]json.RawMessage
|
||||
if err := json.Unmarshal(e.Value, &fields); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, f := range lookFields {
|
||||
if _, held := fields[f]; held {
|
||||
t.Errorf("the bucket's value holds %q, which changes on every look", f)
|
||||
}
|
||||
}
|
||||
for _, f := range []string{"key", "kind", "severity", "resolver", "silenced", "raised", "headline", "needs", "epoch"} {
|
||||
if _, held := fields[f]; !held {
|
||||
t.Errorf("the bucket's value lacks %q", f)
|
||||
}
|
||||
}
|
||||
|
||||
// A change of what it says is written, once, and its event carries the looks.
|
||||
worse := silent("ace")
|
||||
worse.Severity = Urgent
|
||||
if _, err := k.Observe(ctx, worse); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := k.Observe(ctx, worse); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if store.Writes() != 2 {
|
||||
t.Fatalf("a change of severity and one more look wrote the bucket %d times in all, want 2", store.Writes())
|
||||
}
|
||||
said := settled(t, told, 2)
|
||||
if said[1].Change != ChangeSeverity || said[1].Observations != 6 || len(said[1].Evidence) != 6 {
|
||||
t.Fatalf("the severity's event says %s with %d observation(s) and %d evidence line(s)", said[1].Change,
|
||||
said[1].Observations, len(said[1].Evidence))
|
||||
}
|
||||
|
||||
// Its clearing forgets its looks, and its event carries them as they were.
|
||||
if cleared, err := k.Clear(ctx, "machine.ace.silent", "heard"); err != nil || !cleared {
|
||||
t.Fatalf("cleared %v: %v", cleared, err)
|
||||
}
|
||||
if _, kept, _ := store.Looked(ctx, "machine.ace.silent"); kept {
|
||||
t.Fatal("a cleared condition's looks are still kept")
|
||||
}
|
||||
said = settled(t, told, 3)
|
||||
if said[2].Change != ChangeCleared || said[2].Observations != 7 {
|
||||
t.Fatalf("the clearing says %s with %d observation(s)", said[2].Change, said[2].Observations)
|
||||
}
|
||||
}
|
||||
|
||||
// **A condition written before its looks moved out keeps its count**: its value says them, the first
|
||||
// look counts on from there, and writes the value once without them.
|
||||
func TestAConditionWrittenWithItsLooksMovesThemOutOnce(t *testing.T) {
|
||||
k, store, _, c := keeper(t)
|
||||
ctx := t.Context()
|
||||
old := Condition{Key: "machine.ace.silent", Kind: "silent", Subject: Subject{Scope: ScopeMachine, ID: "ace", Machine: "ace"},
|
||||
Severity: Warning, Summary: "ace has not been heard from", Source: "S1", Raised: c.now().Add(-time.Hour),
|
||||
LastObserved: c.now().Add(-time.Minute), Observations: 7003, Count: 1, Resolver: ResolverSelf,
|
||||
Evidence: []Evidence{{At: c.now().Add(-time.Minute), Said: "ace has not been heard from"}}}
|
||||
body, _ := json.Marshal(old)
|
||||
if err := store.Create(ctx, old.Key, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, _, _ := k.Get(ctx, old.Key); got.Observations != 7003 || got.LooksUnread != "" {
|
||||
t.Fatalf("before its first look it reads %d observation(s) (%q)", got.Observations, got.LooksUnread)
|
||||
}
|
||||
for range 3 {
|
||||
if _, err := k.Observe(ctx, silent("ace")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
c.pass(30 * time.Second)
|
||||
}
|
||||
if store.Writes() != 2 {
|
||||
t.Fatalf("three looks wrote the bucket %d times after its creation, want 1", store.Writes()-1)
|
||||
}
|
||||
got, _, _ := k.Get(ctx, old.Key)
|
||||
if got.Observations != 7006 || len(got.Evidence) != 4 {
|
||||
t.Fatalf("it counts %d observation(s) and %d evidence line(s), want 7006 and 4", got.Observations, len(got.Evidence))
|
||||
}
|
||||
}
|
||||
|
||||
// failingLooks is a controller's store that is away.
|
||||
type failingLooks struct{}
|
||||
|
||||
func (failingLooks) KeepLooks(context.Context, string, Looks) error { return errors.New("store away") }
|
||||
func (failingLooks) Looked(context.Context, string) (Looks, bool, error) {
|
||||
return Looks{}, false, errors.New("store away")
|
||||
}
|
||||
func (failingLooks) AllLooks(context.Context) (map[string]Looks, error) {
|
||||
return nil, errors.New("store away")
|
||||
}
|
||||
func (failingLooks) ForgetLooks(context.Context, string) error { return errors.New("store away") }
|
||||
|
||||
// **A condition is raised while the controller's store is away**, and says its looks are unread rather
|
||||
// than none: a store gone is exactly what a condition must still be raised for.
|
||||
func TestLooksThatCannotBeKeptAreSaidNotCountedAsNone(t *testing.T) {
|
||||
store, told := NewInMemory(), &Told{}
|
||||
var said []string
|
||||
k := NewKeeper(t.Context(), Options{Store: store, History: store, Teller: told, Looks: failingLooks{},
|
||||
Say: func(f string, a ...any) { said = append(said, f) }})
|
||||
t.Cleanup(func() { k.Close(context.Background()) })
|
||||
if _, err := k.Observe(t.Context(), silent("ace")); err != nil {
|
||||
t.Fatalf("a condition was not raised for a store away: %v", err)
|
||||
}
|
||||
if _, err := k.Observe(t.Context(), silent("ace")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, open, err := k.Get(t.Context(), "machine.ace.silent")
|
||||
if err != nil || !open || got.LooksUnread == "" {
|
||||
t.Fatalf("open %v, looks unread %q: %v", open, got.LooksUnread, err)
|
||||
}
|
||||
list, err := k.Open(t.Context())
|
||||
if err != nil || len(list) != 1 || list[0].LooksUnread == "" {
|
||||
t.Fatalf("the list reads %+v: %v", list, err)
|
||||
}
|
||||
if e := settled(t, told, 1); e[0].LooksUnread == "" {
|
||||
t.Fatal("the raising says nothing of its looks being unkept")
|
||||
}
|
||||
if len(said) == 0 {
|
||||
t.Fatal("looks that could not be kept were not said")
|
||||
}
|
||||
}
|
||||
@@ -12,10 +12,14 @@ import (
|
||||
// InMemory is a store and a history held in this process: for tests, and for nothing else — a
|
||||
// condition kept here is forgotten by a restart, which is the fault the store exists to remove.
|
||||
type InMemory struct {
|
||||
// MemoryLooks keeps its conditions' looks, as the controller's store does beside the bucket.
|
||||
MemoryLooks
|
||||
mu sync.Mutex
|
||||
values map[string]Entry
|
||||
revision uint64
|
||||
events []Event
|
||||
// writes counts every create, update and delete, so a test can say a look wrote nothing.
|
||||
writes int
|
||||
events []Event
|
||||
// fail, when set, is what every read and write answers: a store that is away. It is set only
|
||||
// through SetFail, under the lock, because the keeper's telling goroutine reads it while a test
|
||||
// takes the store away.
|
||||
@@ -23,7 +27,9 @@ type InMemory struct {
|
||||
}
|
||||
|
||||
// NewInMemory is an empty store.
|
||||
func NewInMemory() *InMemory { return &InMemory{values: map[string]Entry{}} }
|
||||
func NewInMemory() *InMemory {
|
||||
return &InMemory{values: map[string]Entry{}, MemoryLooks: MemoryLooks{looks: map[string]Looks{}}}
|
||||
}
|
||||
|
||||
// SetFail makes every read and write answer err from now on, or none when err is nil.
|
||||
func (m *InMemory) SetFail(err error) {
|
||||
@@ -52,10 +58,18 @@ func (m *InMemory) Create(_ context.Context, key string, value []byte) error {
|
||||
return ErrMoved
|
||||
}
|
||||
m.revision++
|
||||
m.writes++
|
||||
m.values[key] = Entry{Value: value, Revision: m.revision}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Writes is how many creates, updates and deletes the store took.
|
||||
func (m *InMemory) Writes() int {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.writes
|
||||
}
|
||||
|
||||
func (m *InMemory) Update(_ context.Context, key string, value []byte, revision uint64) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
@@ -66,6 +80,7 @@ func (m *InMemory) Update(_ context.Context, key string, value []byte, revision
|
||||
return ErrMoved
|
||||
}
|
||||
m.revision++
|
||||
m.writes++
|
||||
m.values[key] = Entry{Value: value, Revision: m.revision}
|
||||
return nil
|
||||
}
|
||||
@@ -79,6 +94,7 @@ func (m *InMemory) Delete(_ context.Context, key string, revision uint64) error
|
||||
if e, ok := m.values[key]; !ok || e.Revision != revision {
|
||||
return ErrMoved
|
||||
}
|
||||
m.writes++
|
||||
delete(m.values, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -277,11 +277,15 @@ func TestAWriteThatLostTheRaceRedoesItself(t *testing.T) {
|
||||
}
|
||||
}}
|
||||
k.store = racing
|
||||
got, err := k.Observe(ctx, silent("ace"))
|
||||
// An observation that changes what the condition says, so it writes (one that changes nothing writes
|
||||
// nothing, novox/hq issue 420).
|
||||
worse := silent("ace")
|
||||
worse.Severity = Urgent
|
||||
got, err := k.Observe(ctx, worse)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Silenced == nil || got.Observations != 2 {
|
||||
if got.Silenced == nil || got.Observations != 2 || got.Severity != Urgent {
|
||||
t.Fatalf("the observation overwrote the silence: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// The looks of the open conditions (novox/hq issue 420): see migration 0095. How many times each was
|
||||
// observed, when last, and the newest evidence change on every look, so they are kept here rather than in
|
||||
// the conditions bucket, which is written only when what a condition says changes (ADR 0289 §1).
|
||||
|
||||
// ConditionLooks is one open condition's looks. Evidence is kept as the condition says it, newest first.
|
||||
type ConditionLooks struct {
|
||||
Key string
|
||||
Observations int
|
||||
LastObserved time.Time
|
||||
Evidence json.RawMessage
|
||||
}
|
||||
|
||||
// KeepConditionLooks writes a condition's looks, replacing what its row held. Written under the lease: a
|
||||
// controller that may not act keeps no looks.
|
||||
func (i *Inventory) KeepConditionLooks(ctx context.Context, l ConditionLooks) error {
|
||||
if l.Key == "" {
|
||||
return errors.New("a condition's looks name its key")
|
||||
}
|
||||
epoch, err := i.actingEpoch(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
evidence := l.Evidence
|
||||
if len(evidence) == 0 {
|
||||
evidence = json.RawMessage("[]")
|
||||
}
|
||||
_, err = i.store.Pool().Exec(ctx,
|
||||
`insert into condition_looks (key, observations, last_observed, evidence, epoch, kept_at)
|
||||
values ($1, $2, $3, $4, $5, now())
|
||||
on conflict (key) do update set observations = excluded.observations,
|
||||
last_observed = excluded.last_observed, evidence = excluded.evidence, epoch = excluded.epoch,
|
||||
kept_at = excluded.kept_at`,
|
||||
l.Key, l.Observations, l.LastObserved, []byte(evidence), epoch)
|
||||
return err
|
||||
}
|
||||
|
||||
// ConditionLooksOf is one condition's looks; false when none are kept.
|
||||
func (i *Inventory) ConditionLooksOf(ctx context.Context, key string) (ConditionLooks, bool, error) {
|
||||
l := ConditionLooks{Key: key}
|
||||
var evidence []byte
|
||||
err := i.store.Pool().QueryRow(ctx,
|
||||
`select observations, last_observed, evidence from condition_looks where key = $1`, key).
|
||||
Scan(&l.Observations, &l.LastObserved, &evidence)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return ConditionLooks{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return ConditionLooks{}, false, err
|
||||
}
|
||||
l.Evidence = evidence
|
||||
return l, true, nil
|
||||
}
|
||||
|
||||
// AllConditionLooks is every kept condition's looks, by key.
|
||||
func (i *Inventory) AllConditionLooks(ctx context.Context) (map[string]ConditionLooks, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `select key, observations, last_observed, evidence from condition_looks`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string]ConditionLooks{}
|
||||
for rows.Next() {
|
||||
var l ConditionLooks
|
||||
var evidence []byte
|
||||
if err := rows.Scan(&l.Key, &l.Observations, &l.LastObserved, &evidence); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
l.Evidence = evidence
|
||||
out[l.Key] = l
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// ForgetConditionLooks removes a condition's looks: it cleared. Whatever the lease says by now, as a
|
||||
// clearing is: what is not open has nothing to count.
|
||||
func (i *Inventory) ForgetConditionLooks(ctx context.Context, key string) error {
|
||||
_, err := i.store.Pool().Exec(ctx, `delete from condition_looks where key = $1`, key)
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// **A condition's looks are kept in the store, replaced on every look and forgotten when it clears**
|
||||
// (novox/hq issue 420): written under the lease, read back whole.
|
||||
func TestAConditionsLooksAreKeptBesideTheBucket(t *testing.T) {
|
||||
inv := ForTest(t)
|
||||
ctx := t.Context()
|
||||
inv.ActsUnder(func(context.Context) (uint64, error) { return 0, errors.New("the lease is not held") })
|
||||
if err := inv.KeepConditionLooks(ctx, ConditionLooks{Key: "machine.ace.silent", Observations: 1,
|
||||
LastObserved: time.Now()}); err == nil {
|
||||
t.Fatal("looks were kept by a process that may not act")
|
||||
}
|
||||
inv.ActsUnder(func(context.Context) (uint64, error) { return 57, nil })
|
||||
at := time.Now().UTC().Truncate(time.Microsecond)
|
||||
for n := 1; n <= 3; n++ {
|
||||
evidence, _ := json.Marshal([]map[string]any{{"at": at, "said": "ace has not been heard from"}})
|
||||
if err := inv.KeepConditionLooks(ctx, ConditionLooks{Key: "machine.ace.silent", Observations: n,
|
||||
LastObserved: at, Evidence: evidence}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := inv.KeepConditionLooks(ctx, ConditionLooks{Key: "machine.ace.late", Observations: 2,
|
||||
LastObserved: at}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, kept, err := inv.ConditionLooksOf(ctx, "machine.ace.silent")
|
||||
if err != nil || !kept || got.Observations != 3 || !got.LastObserved.Equal(at) {
|
||||
t.Fatalf("read %+v kept %v: %v", got, kept, err)
|
||||
}
|
||||
var evidence []map[string]any
|
||||
if err := json.Unmarshal(got.Evidence, &evidence); err != nil || len(evidence) != 1 {
|
||||
t.Fatalf("evidence %s: %v", got.Evidence, err)
|
||||
}
|
||||
all, err := inv.AllConditionLooks(ctx)
|
||||
if err != nil || len(all) != 2 || all["machine.ace.late"].Observations != 2 ||
|
||||
string(all["machine.ace.late"].Evidence) != "[]" {
|
||||
t.Fatalf("all %+v: %v", all, err)
|
||||
}
|
||||
if err := inv.ForgetConditionLooks(ctx, "machine.ace.silent"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, kept, err := inv.ConditionLooksOf(ctx, "machine.ace.silent"); kept || err != nil {
|
||||
t.Fatalf("a cleared condition's looks kept %v: %v", kept, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
-- The looks of an open condition are kept here, not in the conditions bucket (novox/hq issue 420, ADR 0289
|
||||
-- §1 and §4).
|
||||
--
|
||||
-- On 2026-10-10 the bus had stored 131 647 writes to the conditions bucket in four days while twelve
|
||||
-- conditions were open: every look rewrote a condition's value, because the value carried how many times it
|
||||
-- had been observed, when last and the newest evidence. A watcher heard a write for every look, not for
|
||||
-- every change. The bucket now holds what a condition says, written when that changes; how often and when
|
||||
-- it was last seen, and what each look said, change on every look and are kept in this row, written by the
|
||||
-- serving controller under its lease (`epoch`) and read back by `conditions` and the events it says.
|
||||
--
|
||||
-- One row per open condition, removed when it clears. `evidence` is the newest observations, newest first,
|
||||
-- as the condition's `evidence` says them: [{"at", "said"}].
|
||||
create table condition_looks (
|
||||
key text primary key,
|
||||
observations integer not null check (observations >= 0),
|
||||
last_observed timestamptz not null,
|
||||
evidence jsonb not null default '[]',
|
||||
epoch bigint,
|
||||
kept_at timestamptz not null default now()
|
||||
);
|
||||
Vendored
+2
@@ -12,3 +12,5 @@
|
||||
./cmd/mesh-controller TestNodeShowAndStatusReadTheLastSendFromTheStore
|
||||
./internal/inventory TestAnIssuedValueIsSealedToEveryHolderOnceAndNotAgain
|
||||
./internal/inventory TestAnIssuedValueIsSealedAgainAfterAKeyChangeOrAnAcceptance
|
||||
./internal/inventory TestAConditionsLooksAreKeptBesideTheBucket
|
||||
./cmd/mesh-controller TestTheConditionsVerbReadsTheLooksTheBucketNoLongerHolds
|
||||
|
||||
Reference in New Issue
Block a user