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.
181 lines
4.2 KiB
Go
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...)
|
|
}
|