Files
mesh-controller/internal/conditions/memory.go
T
jochen dcdafd37f0
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…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery delivered
Keep a condition's looks in the store, so a look that changes nothing writes no bucket (issue 420)
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.
2026-10-12 12:11:13 +02:00

181 lines
4.2 KiB
Go

package conditions
import (
"context"
"encoding/json"
"errors"
"sort"
"sync"
"time"
)
// 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
// 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.
fail error
}
// NewInMemory is an empty store.
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) {
m.mu.Lock()
defer m.mu.Unlock()
m.fail = err
}
func (m *InMemory) Get(_ context.Context, key string) (Entry, bool, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.fail != nil {
return Entry{}, false, m.fail
}
e, ok := m.values[key]
return e, ok, nil
}
func (m *InMemory) Create(_ context.Context, key string, value []byte) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.fail != nil {
return m.fail
}
if _, ok := m.values[key]; ok {
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()
if m.fail != nil {
return m.fail
}
if e, ok := m.values[key]; !ok || e.Revision != revision {
return ErrMoved
}
m.revision++
m.writes++
m.values[key] = Entry{Value: value, Revision: m.revision}
return nil
}
func (m *InMemory) Delete(_ context.Context, key string, revision uint64) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.fail != nil {
return m.fail
}
if e, ok := m.values[key]; !ok || e.Revision != revision {
return ErrMoved
}
m.writes++
delete(m.values, key)
return nil
}
func (m *InMemory) All(context.Context) (map[string]Entry, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.fail != nil {
return nil, m.fail
}
out := make(map[string]Entry, len(m.values))
for k, v := range m.values {
out[k] = v
}
return out, nil
}
func (m *InMemory) Append(_ context.Context, e Event) error {
m.mu.Lock()
defer m.mu.Unlock()
if m.fail != nil {
return m.fail
}
m.events = append(m.events, e)
return nil
}
func (m *InMemory) Since(_ context.Context, since time.Time) ([]Event, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.fail != nil {
return nil, m.fail
}
var out []Event
for _, e := range m.events {
if !e.At.Before(since) {
out = append(out, e)
}
}
sort.SliceStable(out, func(i, j int) bool { return out[i].At.Before(out[j].At) })
return out, nil
}
// Told is a teller that remembers what it was told, for tests.
type Told struct {
mu sync.Mutex
Events []Event
Names []string
// fail, when set, is what every publish answers; set only through SetFail, under the lock.
fail error
}
// SetFail makes every publish answer err from now on, or none when err is nil.
func (t *Told) SetFail(err error) {
t.mu.Lock()
defer t.mu.Unlock()
t.fail = err
}
func (t *Told) PublishSeatEvent(_ context.Context, seat, event string, body []byte) error {
t.mu.Lock()
defer t.mu.Unlock()
if t.fail != nil {
return t.fail
}
if seat != Seat {
return errors.New("told under the wrong seat: " + seat)
}
var e Event
if err := json.Unmarshal(body, &e); err != nil {
return err
}
t.Events = append(t.Events, e)
t.Names = append(t.Names, event)
return nil
}
// Said is a copy of what was told so far.
func (t *Told) Said() []Event {
t.mu.Lock()
defer t.mu.Unlock()
return append([]Event(nil), t.Events...)
}