CAT-N3: the keeper creates each template's bus objects under the module's names and says created or refused
Each bus-state entry is planned into its objects (design 25 § 12.1): state a bucket <module>_<local>; events, work queue, log and a bare stream a stream <module>_<local> on mesh.bus.<module>.<local>.>; objects an object store; a ledger its history stream, its state and index buckets and its snapshot stream, one message per subject. The keeper makes what the bus lacks on its own account; a refusal of its grant waits for the next check, a refusal of the entry is said (refused) and raised, cleared by a later create. The comparison of what is held and the orphans are in the same pass (CAT-N4, N5). Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,159 @@
|
||||
package keeper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// JetStream is the bus as the keeper's own account reaches it (Bus): each object a stream, a key-value bucket or an
|
||||
// object store of JetStream, read, created, grown and deleted by its stream.
|
||||
type JetStream struct {
|
||||
JS jetstream.JetStream
|
||||
// Snap takes a snapshot of streams and says where it is kept (bussnap through the keeper's own connection).
|
||||
Snap func(ctx context.Context, streams []string) (string, error)
|
||||
}
|
||||
|
||||
// Held reads the object's stream.
|
||||
func (b JetStream) Held(ctx context.Context, o Object) (Settings, Fill, bool, error) {
|
||||
s, err := b.JS.Stream(ctx, o.Stream())
|
||||
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return Settings{}, Fill{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return Settings{}, Fill{}, false, retryable(err)
|
||||
}
|
||||
info, err := s.Info(ctx)
|
||||
if err != nil {
|
||||
return Settings{}, Fill{}, false, retryable(err)
|
||||
}
|
||||
return settingsOf(o.Kind, info.Config), Fill{Bytes: info.State.Bytes, Messages: info.State.Msgs}, true, nil
|
||||
}
|
||||
|
||||
// settingsOf reads a stream's configuration as the settings an object of kind is asked with.
|
||||
func settingsOf(kind string, c jetstream.StreamConfig) Settings {
|
||||
s := Settings{MaxBytes: c.MaxBytes, MaxAgeSeconds: int64(c.MaxAge / time.Second), Discard: DiscardOld,
|
||||
WorkQueue: c.Retention == jetstream.WorkQueuePolicy, DuplicatesSeconds: int64(c.Duplicates / time.Second)}
|
||||
if c.Discard == jetstream.DiscardNew {
|
||||
s.Discard = DiscardNew
|
||||
}
|
||||
if c.MaxMsgSize > 0 {
|
||||
s.MaxMsgSize = int64(c.MaxMsgSize)
|
||||
}
|
||||
switch kind {
|
||||
case KindBucket:
|
||||
s.History = c.MaxMsgsPerSubject
|
||||
s.KeyTTL = c.AllowMsgTTL
|
||||
s.DuplicatesSeconds = 0
|
||||
case KindObjects:
|
||||
s.DuplicatesSeconds = 0
|
||||
default:
|
||||
if c.MaxMsgsPerSubject > 0 {
|
||||
s.History = c.MaxMsgsPerSubject
|
||||
}
|
||||
if !s.WorkQueue {
|
||||
s.DuplicatesSeconds = 0
|
||||
}
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// Create creates the object; a refusal of the account's grant is ErrRetry: the grants part composes the keeper's grant
|
||||
// to create a module's names from the same accepted manifests, and may not have yet.
|
||||
func (b JetStream) Create(ctx context.Context, o Object) error {
|
||||
var err error
|
||||
switch o.Kind {
|
||||
case KindBucket:
|
||||
_, err = b.JS.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: o.Name, History: uint8(o.Settings.History),
|
||||
TTL: time.Duration(o.Settings.MaxAgeSeconds) * time.Second, MaxBytes: o.Settings.MaxBytes,
|
||||
MaxValueSize: int32(o.Settings.MaxMsgSize), Storage: jetstream.FileStorage, Replicas: 1,
|
||||
LimitMarkerTTL: limitMarker(o.Settings.KeyTTL)})
|
||||
case KindObjects:
|
||||
_, err = b.JS.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: o.Name, MaxBytes: o.Settings.MaxBytes,
|
||||
Storage: jetstream.FileStorage, Replicas: 1})
|
||||
default:
|
||||
_, err = b.JS.CreateStream(ctx, streamConfig(o))
|
||||
}
|
||||
return retryable(err)
|
||||
}
|
||||
|
||||
func limitMarker(on bool) time.Duration {
|
||||
if on {
|
||||
return time.Second
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func streamConfig(o Object) jetstream.StreamConfig {
|
||||
c := jetstream.StreamConfig{Name: o.Name, Subjects: o.Subjects, Retention: jetstream.LimitsPolicy,
|
||||
MaxBytes: o.Settings.MaxBytes, MaxAge: time.Duration(o.Settings.MaxAgeSeconds) * time.Second,
|
||||
Discard: jetstream.DiscardOld, Storage: jetstream.FileStorage, Replicas: 1, MaxMsgs: -1, MaxMsgsPerSubject: -1}
|
||||
if o.Settings.Discard == DiscardNew {
|
||||
c.Discard = jetstream.DiscardNew
|
||||
}
|
||||
if o.Settings.WorkQueue {
|
||||
c.Retention = jetstream.WorkQueuePolicy
|
||||
c.Duplicates = time.Duration(o.Settings.DuplicatesSeconds) * time.Second
|
||||
}
|
||||
if o.Settings.MaxMsgSize > 0 {
|
||||
c.MaxMsgSize = int32(o.Settings.MaxMsgSize)
|
||||
}
|
||||
if o.Settings.History > 0 {
|
||||
c.MaxMsgsPerSubject = o.Settings.History
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// Update brings the held stream to the object's settings, keeping every other part of its configuration.
|
||||
func (b JetStream) Update(ctx context.Context, o Object) error {
|
||||
s, err := b.JS.Stream(ctx, o.Stream())
|
||||
if err != nil {
|
||||
return retryable(err)
|
||||
}
|
||||
info, err := s.Info(ctx)
|
||||
if err != nil {
|
||||
return retryable(err)
|
||||
}
|
||||
c := info.Config
|
||||
c.MaxBytes = o.Settings.MaxBytes
|
||||
c.MaxAge = time.Duration(o.Settings.MaxAgeSeconds) * time.Second
|
||||
if o.Settings.History > 0 {
|
||||
c.MaxMsgsPerSubject = o.Settings.History
|
||||
}
|
||||
_, err = b.JS.UpdateStream(ctx, c)
|
||||
return retryable(err)
|
||||
}
|
||||
|
||||
// Delete deletes the object's stream.
|
||||
func (b JetStream) Delete(ctx context.Context, o Object) error {
|
||||
err := b.JS.DeleteStream(ctx, o.Stream())
|
||||
if errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return nil
|
||||
}
|
||||
return retryable(err)
|
||||
}
|
||||
|
||||
// Snapshot takes the streams' snapshot.
|
||||
func (b JetStream) Snapshot(ctx context.Context, streams []string) (string, error) {
|
||||
if b.Snap == nil {
|
||||
return "", fmt.Errorf("no snapshot is taken here, so nothing is removed")
|
||||
}
|
||||
return b.Snap(ctx, streams)
|
||||
}
|
||||
|
||||
// retryable marks a refusal of the account's permissions, or a bus that answers nothing, as ErrRetry.
|
||||
func retryable(err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
words := strings.ToLower(err.Error())
|
||||
if strings.Contains(words, "permission") || strings.Contains(words, "authorization") || errors.Is(err, context.DeadlineExceeded) ||
|
||||
strings.Contains(words, "no responders") || strings.Contains(words, "timeout") {
|
||||
return fmt.Errorf("%w: %v", ErrRetry, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
@@ -0,0 +1,599 @@
|
||||
package keeper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
|
||||
// Module is the bus module's own name: the publisher of its events and its conditions.
|
||||
const Module = "nats"
|
||||
|
||||
// The keeper's own states (old `state` sections, ADR 0595 §10): what it last heard per module, and the persons'
|
||||
// requests its tools hand it.
|
||||
const (
|
||||
StateAsked = "asked"
|
||||
StateRequests = "requests"
|
||||
// KeyStatus is the keeper's own status in StateAsked, read by its health tool: a key no module can have, since a
|
||||
// module's name holds no underscore.
|
||||
KeyStatus = "_keeper"
|
||||
)
|
||||
|
||||
// The bus module's events (ADR 0595 §6), on its own stream (ADR 0620 §2).
|
||||
const (
|
||||
EventCreated = "created"
|
||||
EventRefused = "refused"
|
||||
EventDrifted = "drifted"
|
||||
EventDriftCleared = "drift-cleared"
|
||||
EventFull = "full"
|
||||
EventFullCleared = "full-cleared"
|
||||
EventOrphaned = "orphaned"
|
||||
EventRecreated = "recreated"
|
||||
EventDropped = "dropped"
|
||||
)
|
||||
|
||||
// Event is the body of every bus module event: the entry, its binding (the names on the bus), the settings asked and
|
||||
// held by bus object, why (refused, drifted), and the indexer's revision of the verdict it acted on.
|
||||
//
|
||||
// Written here until mesh-contracts holds the generated type of the seat mesh-broker's definition (ADR 0613 §1); the
|
||||
// field names are that definition's.
|
||||
type Event struct {
|
||||
V int `json:"v"`
|
||||
Module string `json:"module"`
|
||||
Local string `json:"local"`
|
||||
Storage string `json:"storage"`
|
||||
Template string `json:"template,omitempty"`
|
||||
Binding membership.BusBinding `json:"binding"`
|
||||
Asked map[string]Settings `json:"asked,omitempty"`
|
||||
Held map[string]Settings `json:"held,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
}
|
||||
|
||||
// Accepted is a manifest-accepted the keeper heard, or a module of its first view: the module, its manifest's digest,
|
||||
// the indexer's revision of the verdict (0 when the event does not say it yet), and its bus-state entries whole.
|
||||
type Accepted struct {
|
||||
Module string `json:"module"`
|
||||
Digest string `json:"digest"`
|
||||
Revision uint64 `json:"revision"`
|
||||
BusState []catalogue.BusStateEntry `json:"bus-state"`
|
||||
}
|
||||
|
||||
// FirstView is the indexer's answer of indexer_bus_state: every accepted manifest's bus-state entries, by module.
|
||||
type FirstView struct {
|
||||
Revision uint64 `json:"revision"`
|
||||
Modules []Accepted `json:"modules"`
|
||||
}
|
||||
|
||||
// Asked is what the keeper holds of one module, the value of its key in StateAsked: what its newest accepted manifest
|
||||
// asks for, and every entry the keeper made for it and the bus holds, by local name.
|
||||
type Asked struct {
|
||||
V int `json:"v"`
|
||||
Module string `json:"module"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Entries []catalogue.BusStateEntry `json:"entries"`
|
||||
Made map[string]Made `json:"made,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Made is one entry as the keeper made it: its plan and what stands of it now.
|
||||
type Made struct {
|
||||
Plan Plan `json:"plan"`
|
||||
// Created says every object of it is on the bus and its `created` was said.
|
||||
Created bool `json:"created"`
|
||||
// Refused is why it could not be made, Waiting why it waits to be made again (a refusal the bus may lift, such as a
|
||||
// grant not composed yet).
|
||||
Refused string `json:"refused,omitempty"`
|
||||
Waiting string `json:"waiting,omitempty"`
|
||||
// Drift is how what the bus holds differs from what was asked, in a way that does not grow.
|
||||
Drift string `json:"drift,omitempty"`
|
||||
// Full is whether its fill is past the raise mark, Fill the last fill read, in percent of its cap.
|
||||
Full bool `json:"full,omitempty"`
|
||||
Fill int64 `json:"fill,omitempty"`
|
||||
// Orphaned says no manifest asks for it any more: it goes only by drop.
|
||||
Orphaned bool `json:"orphaned,omitempty"`
|
||||
}
|
||||
|
||||
// Status is the keeper's own state, KeyStatus: what its health tool reads.
|
||||
type Status struct {
|
||||
Started time.Time `json:"started"`
|
||||
// Pass is when the last fill pass ended.
|
||||
Pass time.Time `json:"pass"`
|
||||
// FirstView is the indexer's first view: when asked, when answered and at which revision, or why not.
|
||||
FirstView FirstViewStatus `json:"first-view"`
|
||||
// Heard is when the last manifest-accepted was heard, and Unanswered since when one has waited unhandled.
|
||||
Heard time.Time `json:"heard,omitempty"`
|
||||
Unanswered time.Time `json:"unanswered,omitempty"`
|
||||
}
|
||||
|
||||
// FirstViewStatus is what the keeper holds of its first view.
|
||||
type FirstViewStatus struct {
|
||||
Owner string `json:"owner"`
|
||||
Verb string `json:"verb"`
|
||||
Asked time.Time `json:"asked,omitempty"`
|
||||
Answered time.Time `json:"answered,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Next time.Time `json:"next,omitempty"`
|
||||
}
|
||||
|
||||
// The first view's owner and verb (design 25 § 12.2).
|
||||
const (
|
||||
FirstViewOwner = "mesh-delivery-indexer"
|
||||
FirstViewVerb = "indexer_bus_state"
|
||||
)
|
||||
|
||||
// Bus is the bus as the keeper's own account reaches it.
|
||||
type Bus interface {
|
||||
// Held reads the object's stream: its settings, whether it is there, and its fill (bytes and messages).
|
||||
Held(ctx context.Context, o Object) (Settings, Fill, bool, error)
|
||||
Create(ctx context.Context, o Object) error
|
||||
// Update brings a held object to o's settings: only a difference that grows (Grows), or a recreate's rebuild.
|
||||
Update(ctx context.Context, o Object) error
|
||||
Delete(ctx context.Context, o Object) error
|
||||
// Snapshot takes a snapshot of the streams and says where it is kept.
|
||||
Snapshot(ctx context.Context, streams []string) (string, error)
|
||||
}
|
||||
|
||||
// Fill is how much one object holds: its bytes and messages.
|
||||
type Fill struct {
|
||||
Bytes uint64 `json:"bytes"`
|
||||
Messages uint64 `json:"messages"`
|
||||
}
|
||||
|
||||
// ErrRetry marks a refusal the bus may lift without a change of the manifest (a create the keeper's account is not yet
|
||||
// granted): the entry waits and is made again on the backoff, never said refused.
|
||||
var ErrRetry = errors.New("refused for now")
|
||||
|
||||
// Store is the keeper's own state.
|
||||
type Store interface {
|
||||
Get(state, key string) ([]byte, uint64, bool, error)
|
||||
Put(state, key string, value []byte) (uint64, error)
|
||||
Delete(state, key string) error
|
||||
Keys(state string) ([]string, error)
|
||||
}
|
||||
|
||||
// Keeper is the bus module's keeper.
|
||||
type Keeper struct {
|
||||
Bus Bus
|
||||
Store Store
|
||||
// Emit says one of the bus module's events; Say raises or clears a condition through the condition keeper.
|
||||
Emit func(event string, body Event) error
|
||||
Say func(asks.Condition) error
|
||||
// View asks the indexer's indexer_bus_state.
|
||||
View func() (FirstView, error)
|
||||
// Act records a hand act through the seat mesh-hand-acts; Warrant reads the router's record of an ask.
|
||||
Act func(what, why, cause string) error
|
||||
Warrant func(askID string) (*asks.Ask, *asks.Warrant, error)
|
||||
Now func() time.Time
|
||||
Logf func(string, ...any)
|
||||
// Raise and Clear are the fill marks in percent (ADR 0595 §6): settings of the bus module, 75 and 70 by default.
|
||||
Raise, Clear int64
|
||||
|
||||
mu sync.Mutex
|
||||
status Status
|
||||
viewed bool
|
||||
}
|
||||
|
||||
func (k *Keeper) now() time.Time {
|
||||
if k.Now == nil {
|
||||
return time.Now().UTC()
|
||||
}
|
||||
return k.Now().UTC()
|
||||
}
|
||||
|
||||
func (k *Keeper) logf(format string, a ...any) {
|
||||
if k.Logf != nil {
|
||||
k.Logf(format, a...)
|
||||
}
|
||||
}
|
||||
|
||||
func (k *Keeper) marks() (int64, int64) {
|
||||
raise, clear := k.Raise, k.Clear
|
||||
if raise <= 0 {
|
||||
raise = 75
|
||||
}
|
||||
if clear <= 0 || clear >= raise {
|
||||
clear = raise - 5
|
||||
}
|
||||
return raise, clear
|
||||
}
|
||||
|
||||
// Viewed is whether the first view was answered: before it the keeper creates and orphans nothing.
|
||||
func (k *Keeper) Viewed() bool {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
return k.viewed
|
||||
}
|
||||
|
||||
// Status is the keeper's status now.
|
||||
func (k *Keeper) Status() Status {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
return k.status
|
||||
}
|
||||
|
||||
func (k *Keeper) setStatus(f func(*Status)) {
|
||||
k.mu.Lock()
|
||||
if k.status.Started.IsZero() {
|
||||
k.status.Started = k.now()
|
||||
}
|
||||
k.status.FirstView.Owner, k.status.FirstView.Verb = FirstViewOwner, FirstViewVerb
|
||||
f(&k.status)
|
||||
s := k.status
|
||||
k.mu.Unlock()
|
||||
if k.Store != nil {
|
||||
raw, _ := json.Marshal(s)
|
||||
if _, err := k.Store.Put(StateAsked, KeyStatus, raw); err != nil {
|
||||
k.logf("the keeper's status cannot be written to its own %s: %v", StateAsked, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// AskFirstView asks the indexer's first view once and applies it (design 25 § 12.2): every module it answers is
|
||||
// applied as an accepted verdict, a module the keeper holds and the view leaves out is kept (a set-aside manifest is
|
||||
// missing from it, never removed), and then check runs over what is held. It answers the error and when to ask again,
|
||||
// on the backoff from 10 s doubling to 5 min; the keeper's health is unhealthy meanwhile, naming the verb.
|
||||
func (k *Keeper) AskFirstView(wait time.Duration) (time.Duration, error) {
|
||||
at := k.now()
|
||||
k.setStatus(func(s *Status) { s.FirstView.Asked = at })
|
||||
view, err := k.View()
|
||||
if err != nil {
|
||||
next := wait * 2
|
||||
if wait <= 0 {
|
||||
next = 10 * time.Second
|
||||
}
|
||||
if next > 5*time.Minute {
|
||||
next = 5 * time.Minute
|
||||
}
|
||||
k.setStatus(func(s *Status) {
|
||||
s.FirstView.Error = fmt.Sprintf("%s.%s did not answer: %v", FirstViewOwner, FirstViewVerb, err)
|
||||
s.FirstView.Next = at.Add(next)
|
||||
})
|
||||
return next, err
|
||||
}
|
||||
k.mu.Lock()
|
||||
k.viewed = true
|
||||
k.mu.Unlock()
|
||||
k.setStatus(func(s *Status) {
|
||||
s.FirstView.Answered, s.FirstView.Revision, s.FirstView.Error, s.FirstView.Next = k.now(), view.Revision, "", time.Time{}
|
||||
})
|
||||
for _, a := range view.Modules {
|
||||
k.Heard(a)
|
||||
}
|
||||
k.CheckAll()
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// Heard applies one accepted verdict (manifest-accepted, or a module of the first view): applied only when newer than
|
||||
// what the keeper holds for that module by the indexer's revision alone, so a replay never takes an entry back (a
|
||||
// verdict that says no revision is applied when its digest differs). Each entry asked is made where the bus lacks it;
|
||||
// each entry made that is no longer asked is an orphan. An empty verdict ("this module asks nothing now") ends the
|
||||
// module's key once nothing it made is held.
|
||||
func (k *Keeper) Heard(a Accepted) {
|
||||
k.setStatus(func(s *Status) { s.Heard = k.now() })
|
||||
if !k.Viewed() {
|
||||
k.logf("manifest-accepted of %s heard before the first view: kept on the consumer until it is answered", a.Module)
|
||||
return
|
||||
}
|
||||
held, _, found := k.asked(a.Module)
|
||||
if found && a.Revision != 0 && a.Revision <= held.Revision {
|
||||
k.logf("manifest-accepted of %s at revision %d is not newer than the %d held: not applied", a.Module, a.Revision, held.Revision)
|
||||
return
|
||||
}
|
||||
if found && a.Revision == 0 && a.Digest != "" && a.Digest == held.Digest {
|
||||
return
|
||||
}
|
||||
if !found {
|
||||
held = Asked{V: 1, Module: a.Module}
|
||||
}
|
||||
if held.Made == nil {
|
||||
held.Made = map[string]Made{}
|
||||
}
|
||||
held.Digest, held.Entries, held.At = a.Digest, a.BusState, k.now()
|
||||
if a.Revision > held.Revision {
|
||||
held.Revision = a.Revision
|
||||
}
|
||||
asked := map[string]bool{}
|
||||
for _, e := range a.BusState {
|
||||
asked[e.Name] = true
|
||||
k.makeEntry(&held, e)
|
||||
}
|
||||
for _, local := range sortedKeys(held.Made) {
|
||||
m := held.Made[local]
|
||||
if asked[local] || m.Orphaned {
|
||||
continue
|
||||
}
|
||||
if !m.Created {
|
||||
// Never made: nothing on the bus to orphan.
|
||||
delete(held.Made, local)
|
||||
k.clear(conditionRefused, a.Module, local)
|
||||
continue
|
||||
}
|
||||
m.Orphaned = true
|
||||
held.Made[local] = m
|
||||
k.emit(EventOrphaned, m.Plan, held.Revision, "no accepted manifest of "+a.Module+" asks for it now", nil, nil)
|
||||
k.raise(conditionOrphaned, a.Module, local, asks.SeverityWarning, "Kept data no manifest asks for",
|
||||
"the operator's drop of it, or a manifest asking for it again",
|
||||
fmt.Sprintf("%s of %s is on the bus and no accepted manifest asks for it: it goes only by the bus module's drop, on your warrant (nats_drop), or a manifest asking for it again keeps it", BusName(a.Module, local), a.Module))
|
||||
}
|
||||
k.write(held)
|
||||
}
|
||||
|
||||
// makeEntry makes one asked entry where the bus lacks it, compares it where it holds it, and says what it did.
|
||||
func (k *Keeper) makeEntry(held *Asked, e catalogue.BusStateEntry) {
|
||||
module := held.Module
|
||||
plan, err := PlanOf(module, e)
|
||||
m := held.Made[e.Name]
|
||||
if err != nil {
|
||||
if m.Refused != err.Error() {
|
||||
k.emit(EventRefused, Plan{Module: module, Local: e.Name, Storage: e.Storage, Template: e.Template}, held.Revision, err.Error(), nil, nil)
|
||||
k.raise(conditionRefused, module, e.Name, asks.SeverityWarning, "The bus module refused to create kept data",
|
||||
"a manifest that asks for it in a form the bus module makes",
|
||||
fmt.Sprintf("%s of %s was not created: %v. The module gets no binding for it and fails at its start, naming it", e.Name, module, err))
|
||||
}
|
||||
m.Refused, m.Waiting = err.Error(), ""
|
||||
held.Made[e.Name] = m
|
||||
return
|
||||
}
|
||||
wasOrphan := m.Orphaned
|
||||
m.Plan, m.Orphaned = plan, false
|
||||
if wasOrphan {
|
||||
k.clear(conditionOrphaned, module, e.Name)
|
||||
}
|
||||
k.ensure(held, &m)
|
||||
held.Made[e.Name] = m
|
||||
}
|
||||
|
||||
// ensure brings one planned entry onto the bus: each object the bus lacks created, a held one compared. A difference
|
||||
// that only grows is applied; any other is a drift, said and never fixed quietly.
|
||||
func (k *Keeper) ensure(held *Asked, m *Made) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
asked, holds := map[string]Settings{}, map[string]Settings{}
|
||||
var drifts []string
|
||||
for _, o := range m.Plan.Objects {
|
||||
asked[o.Name] = o.Settings
|
||||
s, _, found, err := k.Bus.Held(ctx, o)
|
||||
if err != nil {
|
||||
m.Waiting = fmt.Sprintf("%s cannot be read: %v", o.Stream(), err)
|
||||
return
|
||||
}
|
||||
if !found {
|
||||
if err := k.Bus.Create(ctx, o); err != nil {
|
||||
if errors.Is(err, ErrRetry) {
|
||||
m.Waiting = fmt.Sprintf("%s cannot be created yet: %v", o.Stream(), err)
|
||||
return
|
||||
}
|
||||
m.Refused, m.Waiting = fmt.Sprintf("%s: %v", o.Stream(), err), ""
|
||||
k.emit(EventRefused, m.Plan, held.Revision, m.Refused, asked, nil)
|
||||
k.raise(conditionRefused, held.Module, m.Plan.Local, asks.SeverityWarning, "The bus module refused to create kept data",
|
||||
"a look at why the bus refused it",
|
||||
fmt.Sprintf("%s of %s was not created: %s. The module gets no binding for it and fails at its start, naming it", m.Plan.Local, held.Module, m.Refused))
|
||||
return
|
||||
}
|
||||
holds[o.Name] = o.Settings
|
||||
continue
|
||||
}
|
||||
holds[o.Name] = s
|
||||
if Differs(o.Settings, s) == "" {
|
||||
continue
|
||||
}
|
||||
if Grows(o.Settings, s) {
|
||||
if err := k.Bus.Update(ctx, o); err != nil {
|
||||
m.Waiting = fmt.Sprintf("%s's larger settings cannot be applied yet: %v", o.Stream(), err)
|
||||
return
|
||||
}
|
||||
holds[o.Name] = o.Settings
|
||||
continue
|
||||
}
|
||||
drifts = append(drifts, o.Stream()+": "+Differs(o.Settings, s))
|
||||
}
|
||||
wasRefused := m.Refused != ""
|
||||
m.Waiting, m.Refused = "", ""
|
||||
if wasRefused {
|
||||
k.clear(conditionRefused, held.Module, m.Plan.Local)
|
||||
}
|
||||
if !m.Created {
|
||||
m.Created = true
|
||||
k.emit(EventCreated, m.Plan, held.Revision, "", asked, holds)
|
||||
}
|
||||
drift := strings.Join(drifts, "; ")
|
||||
switch {
|
||||
case drift != "" && drift != m.Drift:
|
||||
m.Drift = drift
|
||||
k.emit(EventDrifted, m.Plan, held.Revision, drift, asked, holds)
|
||||
k.raise(conditionDrifted, held.Module, m.Plan.Local, asks.SeverityWarning, "Kept data differs from what was asked",
|
||||
"a decision: change the manifest back, or the bus module's recreate on your warrant",
|
||||
fmt.Sprintf("%s of %s differs from what its manifest asks, in a way that does not only grow: %s", m.Plan.Local, held.Module, drift))
|
||||
case drift == "" && m.Drift != "":
|
||||
m.Drift = ""
|
||||
k.emit(EventDriftCleared, m.Plan, held.Revision, "", asked, holds)
|
||||
k.clear(conditionDrifted, held.Module, m.Plan.Local)
|
||||
}
|
||||
}
|
||||
|
||||
// CheckAll compares every entry the keeper holds with the bus, making what is missing (design 25 § 12.2's check at
|
||||
// start) and making again what waits; it answers what it found, one line per entry.
|
||||
func (k *Keeper) CheckAll() []CheckLine {
|
||||
var out []CheckLine
|
||||
keys, err := k.Store.Keys(StateAsked)
|
||||
if err != nil {
|
||||
return []CheckLine{{Why: "the keeper's own " + StateAsked + " cannot be read: " + err.Error()}}
|
||||
}
|
||||
for _, module := range keys {
|
||||
if module == KeyStatus {
|
||||
continue
|
||||
}
|
||||
held, _, found := k.asked(module)
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
for _, local := range sortedKeys(held.Made) {
|
||||
m := held.Made[local]
|
||||
if m.Orphaned || m.Plan.Local == "" {
|
||||
out = append(out, lineOf(module, local, m))
|
||||
continue
|
||||
}
|
||||
for _, e := range held.Entries {
|
||||
if e.Name == local {
|
||||
k.makeEntry(&held, e)
|
||||
}
|
||||
}
|
||||
out = append(out, lineOf(module, local, held.Made[local]))
|
||||
}
|
||||
k.write(held)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// CheckLine is one entry as check answers it.
|
||||
type CheckLine struct {
|
||||
Module string `json:"module,omitempty"`
|
||||
Local string `json:"local,omitempty"`
|
||||
Bus string `json:"bus,omitempty"`
|
||||
Created bool `json:"created"`
|
||||
Refused string `json:"refused,omitempty"`
|
||||
Waiting string `json:"waiting,omitempty"`
|
||||
Drift string `json:"drift,omitempty"`
|
||||
Orphaned bool `json:"orphaned,omitempty"`
|
||||
Full bool `json:"full,omitempty"`
|
||||
Fill int64 `json:"fill-percent,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
}
|
||||
|
||||
func lineOf(module, local string, m Made) CheckLine {
|
||||
return CheckLine{Module: module, Local: local, Bus: BusName(module, local), Created: m.Created, Refused: m.Refused,
|
||||
Waiting: m.Waiting, Drift: m.Drift, Orphaned: m.Orphaned, Full: m.Full, Fill: m.Fill}
|
||||
}
|
||||
|
||||
// Waiting is every entry that waits to be made again, `<module> <local>: <why>`: what the backoff retries.
|
||||
func (k *Keeper) Waiting() []string {
|
||||
var out []string
|
||||
keys, _ := k.Store.Keys(StateAsked)
|
||||
for _, module := range keys {
|
||||
held, _, found := k.asked(module)
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
for _, local := range sortedKeys(held.Made) {
|
||||
if w := held.Made[local].Waiting; w != "" {
|
||||
out = append(out, module+" "+local+": "+w)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// The kinds of condition the keeper raises, each naming the module and the entry (design 25 § 12.3).
|
||||
const (
|
||||
conditionFull = "full"
|
||||
conditionDrifted = "drifted"
|
||||
conditionOrphaned = "orphaned"
|
||||
conditionRefused = "refused"
|
||||
)
|
||||
|
||||
// ConditionKey is the key of one of the keeper's conditions: `nats.<kind>.<module>_<local>`.
|
||||
func ConditionKey(kind, module, local string) string {
|
||||
return Module + "." + kind + "." + BusName(module, local)
|
||||
}
|
||||
|
||||
func (k *Keeper) raise(kind, module, local, severity, headline, needs, explanation string) {
|
||||
if k.Say == nil {
|
||||
return
|
||||
}
|
||||
c := asks.Condition{Key: ConditionKey(kind, module, local), Subject: asks.Scope{Scope: "module", ID: module},
|
||||
State: asks.StateRaised, Severity: severity, Headline: headline, Needs: needs,
|
||||
Summary: explanation,
|
||||
Explanation: asks.NeedsYou + needs + ". " + explanation,
|
||||
Resolved: "cleared by the bus module once it no longer holds"}
|
||||
if err := k.Say(c); err != nil {
|
||||
k.logf("the condition %s cannot be raised: %v", c.Key, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (k *Keeper) clear(kind, module, local string) {
|
||||
if k.Say == nil {
|
||||
return
|
||||
}
|
||||
c := asks.Condition{Key: ConditionKey(kind, module, local), Subject: asks.Scope{Scope: "module", ID: module}, State: asks.StateCleared}
|
||||
if err := k.Say(c); err != nil {
|
||||
k.logf("the condition %s cannot be cleared: %v", c.Key, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (k *Keeper) emit(event string, p Plan, revision uint64, why string, asked, held map[string]Settings) {
|
||||
if k.Emit == nil {
|
||||
return
|
||||
}
|
||||
body := Event{V: 1, Module: p.Module, Local: p.Local, Storage: p.Storage, Template: p.Template, Binding: p.Binding,
|
||||
Asked: asked, Held: held, Why: why, Revision: revision}
|
||||
if event == EventDropped {
|
||||
// A dropped entry's names only: the binding to remove.
|
||||
body.Binding.Types = nil
|
||||
}
|
||||
if err := k.Emit(event, body); err != nil {
|
||||
k.logf("the event %s of %s %s could not be said: %v", event, p.Module, p.Local, err)
|
||||
}
|
||||
}
|
||||
|
||||
func (k *Keeper) asked(module string) (Asked, uint64, bool) {
|
||||
raw, rev, found, err := k.Store.Get(StateAsked, module)
|
||||
if err != nil || !found {
|
||||
return Asked{}, 0, false
|
||||
}
|
||||
var a Asked
|
||||
if json.Unmarshal(raw, &a) != nil {
|
||||
return Asked{}, 0, false
|
||||
}
|
||||
return a, rev, true
|
||||
}
|
||||
|
||||
// write keeps a module's key, or removes it when its manifest asks for nothing and nothing it made is held: a key goes
|
||||
// only so, never because a module is missing from a first view.
|
||||
func (k *Keeper) write(a Asked) {
|
||||
if len(a.Entries) == 0 && len(a.Made) == 0 {
|
||||
if err := k.Store.Delete(StateAsked, a.Module); err != nil {
|
||||
k.logf("the key of %s cannot be removed from its own %s: %v", a.Module, StateAsked, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
raw, _ := json.Marshal(a)
|
||||
if _, err := k.Store.Put(StateAsked, a.Module, raw); err != nil {
|
||||
k.logf("the key of %s cannot be written to its own %s: %v", a.Module, StateAsked, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Held is every module the keeper holds, by name: what list, show and orphans answer from.
|
||||
func (k *Keeper) Held() map[string]Asked {
|
||||
out := map[string]Asked{}
|
||||
keys, _ := k.Store.Keys(StateAsked)
|
||||
for _, module := range keys {
|
||||
if module == KeyStatus {
|
||||
continue
|
||||
}
|
||||
if a, _, found := k.asked(module); found {
|
||||
out[module] = a
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func sortedKeys[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for key := range m {
|
||||
out = append(out, key)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,343 @@
|
||||
package keeper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// fakeBus holds objects by stream, as a bus would, and refuses what a test tells it to.
|
||||
type fakeBus struct {
|
||||
mu sync.Mutex
|
||||
held map[string]Settings
|
||||
fill map[string]uint64
|
||||
refuse map[string]error
|
||||
created []string
|
||||
updated []string
|
||||
deleted []string
|
||||
snaps [][]string
|
||||
}
|
||||
|
||||
func newFakeBus() *fakeBus {
|
||||
return &fakeBus{held: map[string]Settings{}, fill: map[string]uint64{}, refuse: map[string]error{}}
|
||||
}
|
||||
|
||||
func (b *fakeBus) Held(_ context.Context, o Object) (Settings, Fill, bool, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
s, ok := b.held[o.Stream()]
|
||||
return s, Fill{Bytes: b.fill[o.Stream()]}, ok, nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) Create(_ context.Context, o Object) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if err := b.refuse[o.Stream()]; err != nil {
|
||||
return err
|
||||
}
|
||||
b.held[o.Stream()] = o.Settings
|
||||
b.created = append(b.created, o.Stream())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) Update(_ context.Context, o Object) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.held[o.Stream()] = o.Settings
|
||||
b.updated = append(b.updated, o.Stream())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) Delete(_ context.Context, o Object) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
delete(b.held, o.Stream())
|
||||
b.deleted = append(b.deleted, o.Stream())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *fakeBus) Snapshot(_ context.Context, streams []string) (string, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.snaps = append(b.snaps, streams)
|
||||
return "/snapshots/" + strings.Join(streams, "+") + ".tar", nil
|
||||
}
|
||||
|
||||
// memStore is a key-value state in memory.
|
||||
type memStore struct {
|
||||
mu sync.Mutex
|
||||
kv map[string][]byte
|
||||
revs map[string]uint64
|
||||
rev uint64
|
||||
}
|
||||
|
||||
func newMemStore() *memStore { return &memStore{kv: map[string][]byte{}, revs: map[string]uint64{}} }
|
||||
|
||||
func (s *memStore) Get(state, key string) ([]byte, uint64, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
v, ok := s.kv[state+"/"+key]
|
||||
return v, s.revs[state+"/"+key], ok, nil
|
||||
}
|
||||
|
||||
func (s *memStore) Put(state, key string, value []byte) (uint64, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.rev++
|
||||
s.kv[state+"/"+key], s.revs[state+"/"+key] = value, s.rev
|
||||
return s.rev, nil
|
||||
}
|
||||
|
||||
func (s *memStore) Delete(state, key string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
delete(s.kv, state+"/"+key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) Keys(state string) ([]string, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var out []string
|
||||
for k := range s.kv {
|
||||
if st, key, _ := strings.Cut(k, "/"); st == state {
|
||||
out = append(out, key)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// said is what a keeper said: its events and its conditions, in order.
|
||||
type said struct {
|
||||
mu sync.Mutex
|
||||
events []string
|
||||
bodies []Event
|
||||
conds []string
|
||||
}
|
||||
|
||||
func (s *said) emit(event string, body Event) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.events = append(s.events, event+" "+body.Module+" "+body.Local)
|
||||
s.bodies = append(s.bodies, body)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *said) say(c asks.Condition) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.conds = append(s.conds, c.State+" "+c.Key)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *said) reset() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.events, s.bodies, s.conds = nil, nil, nil
|
||||
}
|
||||
|
||||
func newKeeper(view FirstView) (*Keeper, *fakeBus, *memStore, *said) {
|
||||
bus, store, out := newFakeBus(), newMemStore(), &said{}
|
||||
at := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
|
||||
k := &Keeper{Bus: bus, Store: store, Emit: out.emit, Say: out.say,
|
||||
View: func() (FirstView, error) { return view, nil },
|
||||
Now: func() time.Time { return at }}
|
||||
return k, bus, store, out
|
||||
}
|
||||
|
||||
func entry(name, storage, template string, options map[string]string) catalogue.BusStateEntry {
|
||||
e := catalogue.BusStateEntry{Name: name, Storage: storage, Template: template}
|
||||
if len(options) > 0 {
|
||||
e.Options = map[string]json.RawMessage{}
|
||||
for k, v := range options {
|
||||
e.Options[k] = json.RawMessage(v)
|
||||
}
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// What one entry is on the bus (design 25 § 12.1, CAT-N3): each template's objects, named `<module>_<local>`, its
|
||||
// binding, and the fixed settings no option changes; and the entries the keeper refuses itself, naming why.
|
||||
func TestEachTemplateIsPlannedUnderTheModulesNames(t *testing.T) {
|
||||
ledger := entry("records", catalogue.StorageLedger, catalogue.TemplateLedger, map[string]string{"max-mib": "16"})
|
||||
ledger.Types = []catalogue.LedgerType{{Name: "decision", Indexes: []catalogue.LedgerIndex{{Name: "by-topic", On: []string{"topic"}}}}}
|
||||
bare := entry("raw", catalogue.StorageStream, "", nil)
|
||||
bare.Configuration = map[string]json.RawMessage{"when-full": json.RawMessage(`"drop-oldest"`), "max-mib": json.RawMessage("8"), "max-age-days": json.RawMessage("3")}
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
entry catalogue.BusStateEntry
|
||||
objects string
|
||||
binding string
|
||||
refused string
|
||||
}{
|
||||
{name: "state", entry: entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, map[string]string{"history": "5"}),
|
||||
objects: "key-value mesh-knowledge_drafts {64 MiB, age 0, history 5, new, size 262144}",
|
||||
binding: "bucket mesh-knowledge_drafts"},
|
||||
{name: "events, default age and cap", entry: entry("events", catalogue.StorageStream, catalogue.TemplateEvents, nil),
|
||||
objects: "stream mesh-knowledge_events [mesh.bus.mesh-knowledge.events.>] {1024 MiB, age 604800, history 0, old, size 0}",
|
||||
binding: "stream mesh-knowledge_events on mesh.bus.mesh-knowledge.events"},
|
||||
{name: "work queue", entry: entry("jobs", catalogue.StorageStream, catalogue.TemplateWorkQueue, map[string]string{"max-mib": "4", "duplicates": "120"}),
|
||||
objects: "stream mesh-knowledge_jobs [mesh.bus.mesh-knowledge.jobs.>] {4 MiB, age 0, history 0, new, size 0, work-queue, duplicates 120}",
|
||||
binding: "stream mesh-knowledge_jobs on mesh.bus.mesh-knowledge.jobs"},
|
||||
{name: "log", entry: entry("log", catalogue.StorageStream, catalogue.TemplateLog, nil),
|
||||
objects: "stream mesh-knowledge_log [mesh.bus.mesh-knowledge.log.>] {1024 MiB, age 0, history 0, new, size 262144}",
|
||||
binding: "stream mesh-knowledge_log on mesh.bus.mesh-knowledge.log"},
|
||||
{name: "objects", entry: entry("files", catalogue.StorageObjectStore, catalogue.TemplateObjects, map[string]string{"max-mib": "32"}),
|
||||
objects: "object-store mesh-knowledge_files {32 MiB, age 0, history 0, new, size 0}",
|
||||
binding: "objects mesh-knowledge_files"},
|
||||
{name: "ledger", entry: ledger,
|
||||
objects: "stream mesh-knowledge_records [mesh.ledger.mesh-knowledge.records.history.>] {16 MiB, age 0, history 0, new, size 262144}; " +
|
||||
"key-value mesh-knowledge_records_state {64 MiB, age 0, history 1, new, size 262144}; " +
|
||||
"key-value mesh-knowledge_records_index-by-topic {64 MiB, age 0, history 1, new, size 262144}; " +
|
||||
"stream mesh-knowledge_records_snapshot [mesh.ledger.mesh-knowledge.records.snapshot.>] {16 MiB, age 0, history 1, old, size 0}",
|
||||
binding: "history mesh-knowledge_records on mesh.ledger.mesh-knowledge.records.history, state mesh-knowledge_records_state, indexes map[by-topic:mesh-knowledge_records_index-by-topic], snapshot mesh.ledger.mesh-knowledge.records.snapshot"},
|
||||
{name: "bare stream", entry: bare,
|
||||
objects: "stream mesh-knowledge_raw [mesh.bus.mesh-knowledge.raw.>] {8 MiB, age 259200, history 0, old, size 0}",
|
||||
binding: "stream mesh-knowledge_raw on mesh.bus.mesh-knowledge.raw"},
|
||||
{name: "a local name holding the separator", entry: entry("my_drafts", catalogue.StorageKeyValue, catalogue.TemplateState, nil),
|
||||
refused: `the local name "my_drafts" is empty or holds "_"`},
|
||||
{name: "objects with no cap", entry: entry("files", catalogue.StorageObjectStore, catalogue.TemplateObjects, nil),
|
||||
refused: "gives no max-mib"},
|
||||
{name: "a bare stream with no when-full", entry: entry("raw", catalogue.StorageStream, "", nil),
|
||||
refused: "must give when-full"},
|
||||
{name: "a template on the wrong storage type", entry: entry("x", catalogue.StorageKeyValue, catalogue.TemplateLog, nil),
|
||||
refused: "is none the bus module makes"},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
p, err := PlanOf("mesh-knowledge", row.entry)
|
||||
if row.refused != "" {
|
||||
if err == nil || !strings.Contains(err.Error(), row.refused) {
|
||||
t.Fatalf("planned %+v (%v): refused, naming %q", p, err, row.refused)
|
||||
}
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var objs []string
|
||||
for _, o := range p.Objects {
|
||||
s := o.Settings
|
||||
w := fmt.Sprintf("%s %s", o.Kind, o.Name)
|
||||
if len(o.Subjects) > 0 {
|
||||
w += fmt.Sprintf(" %v", o.Subjects)
|
||||
}
|
||||
w += fmt.Sprintf(" {%d MiB, age %d, history %d, %s, size %d", s.MaxBytes/MiB, s.MaxAgeSeconds, s.History, s.Discard, s.MaxMsgSize)
|
||||
if s.WorkQueue {
|
||||
w += fmt.Sprintf(", work-queue, duplicates %d", s.DuplicatesSeconds)
|
||||
}
|
||||
objs = append(objs, w+"}")
|
||||
}
|
||||
if got := strings.Join(objs, "; "); got != row.objects {
|
||||
t.Errorf("objects\n got %s\nwant %s", got, row.objects)
|
||||
}
|
||||
b := p.Binding
|
||||
var binding string
|
||||
switch {
|
||||
case b.History != "":
|
||||
binding = fmt.Sprintf("history %s on %s, state %s, indexes %v, snapshot %s", b.History, b.Subject, b.State, b.Indexes, b.Snapshot)
|
||||
case b.Stream != "":
|
||||
binding = fmt.Sprintf("stream %s on %s", b.Stream, b.Subject)
|
||||
case b.Bucket != "":
|
||||
binding = "bucket " + b.Bucket
|
||||
case b.ObjectStore != "":
|
||||
binding = "objects " + b.ObjectStore
|
||||
}
|
||||
if binding != row.binding || b.Local != row.entry.Name || b.Storage != row.entry.Storage {
|
||||
t.Errorf("binding %s (%+v), want %s", binding, b, row.binding)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The keeper makes what an accepted verdict asks for, once, and says it (design 25 § 12.2, ADR 0595 §4; CAT-N2, N3):
|
||||
// nothing before its first view; a replay not newer than what it holds is not applied; a refusal of its grant waits and
|
||||
// is made on the backoff, never said refused; a refusal of the entry is said and raised, and cleared by a later make.
|
||||
func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
|
||||
drafts := entry("drafts", catalogue.StorageKeyValue, catalogue.TemplateState, nil)
|
||||
log := entry("log", catalogue.StorageStream, catalogue.TemplateLog, nil)
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
run func(t *testing.T, k *Keeper, bus *fakeBus, out *said)
|
||||
}{
|
||||
{"created once, binding said, nothing before the first view", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
|
||||
k.mu.Lock()
|
||||
k.viewed = false
|
||||
k.mu.Unlock()
|
||||
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts}})
|
||||
if len(bus.created) != 0 {
|
||||
t.Fatalf("created %v before the first view", bus.created)
|
||||
}
|
||||
if _, err := k.AskFirstView(0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
|
||||
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 3, BusState: []catalogue.BusStateEntry{drafts, log}})
|
||||
if want := "[KV_m_drafts m_log]"; fmt.Sprint(bus.created) != want {
|
||||
t.Errorf("created %v, want %s", bus.created, want)
|
||||
}
|
||||
if want := "[created m drafts created m log]"; fmt.Sprint(out.events) != want {
|
||||
t.Errorf("said %v, want %s", out.events, want)
|
||||
}
|
||||
if b := out.bodies[1]; b.Binding.Stream != "m_log" || b.Binding.Subject != "mesh.bus.m.log" || b.Revision != 3 || b.Asked["m_log"].MaxBytes == 0 {
|
||||
t.Errorf("created carries %+v: the binding, the revision and the settings asked", b)
|
||||
}
|
||||
}},
|
||||
{"a replay not newer is not applied", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
|
||||
k.Heard(Accepted{Module: "m", Digest: "d2", Revision: 9, BusState: []catalogue.BusStateEntry{drafts}})
|
||||
k.Heard(Accepted{Module: "m", Digest: "d1", Revision: 4, BusState: nil})
|
||||
if a := k.Held()["m"]; a.Revision != 9 || len(a.Entries) != 1 || a.Made["drafts"].Orphaned {
|
||||
t.Errorf("held %+v: the replay at 4 took back what 9 asked", a)
|
||||
}
|
||||
}},
|
||||
{"a grant not composed yet waits, then is made", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
|
||||
bus.refuse["m_log"] = fmt.Errorf("%w: permissions violation", ErrRetry)
|
||||
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})
|
||||
if w := k.Waiting(); len(w) != 1 || len(out.events) != 0 {
|
||||
t.Fatalf("waiting %v, said %v: waits, says nothing", w, out.events)
|
||||
}
|
||||
delete(bus.refuse, "m_log")
|
||||
k.CheckAll()
|
||||
if len(k.Waiting()) != 0 || fmt.Sprint(out.events) != "[created m log]" {
|
||||
t.Errorf("waiting %v, said %v: made at the next check", k.Waiting(), out.events)
|
||||
}
|
||||
}},
|
||||
{"a refused create is said and raised, cleared by a later create", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
|
||||
bus.refuse["KV_m_drafts"] = errors.New("insufficient resources")
|
||||
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}})
|
||||
if fmt.Sprint(out.events) != "[refused m drafts]" || fmt.Sprint(out.conds) != "[raised nats.refused.m_drafts]" ||
|
||||
!strings.Contains(out.bodies[0].Why, "insufficient resources") {
|
||||
t.Fatalf("said %v %v %+v", out.events, out.conds, out.bodies)
|
||||
}
|
||||
delete(bus.refuse, "KV_m_drafts")
|
||||
out.reset()
|
||||
k.CheckAll()
|
||||
if fmt.Sprint(out.events) != "[created m drafts]" || fmt.Sprint(out.conds) != "[cleared nats.refused.m_drafts]" {
|
||||
t.Errorf("said %v %v: created, the refusal cleared", out.events, out.conds)
|
||||
}
|
||||
}},
|
||||
{"an entry the keeper cannot plan is refused, naming why", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
|
||||
k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{entry("a_b", catalogue.StorageKeyValue, catalogue.TemplateState, nil)}})
|
||||
if len(bus.created) != 0 || fmt.Sprint(out.events) != "[refused m a_b]" {
|
||||
t.Errorf("created %v, said %v", bus.created, out.events)
|
||||
}
|
||||
}},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
k, bus, _, out := newKeeper(FirstView{Revision: 1})
|
||||
if _, err := k.AskFirstView(0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
row.run(t, k, bus, out)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,349 @@
|
||||
// Package keeper is the bus module's keeper (novox/hq design 25 § 12.2-12.3, ADR 0595 §4-§6): it creates the bus state
|
||||
// every accepted manifest asks for in `requires`, compares what the bus holds with what was asked, finds orphans and
|
||||
// reads the fill of what refuses when full, and says each on the bus module's own events. It reads nothing of another
|
||||
// part's state: what to create comes on the indexer's manifest-accepted, and its first view from the indexer's tool
|
||||
// indexer_bus_state. Everything it touches it reaches through the interfaces in keeper.go; this file is pure.
|
||||
package keeper
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
|
||||
// The kinds of bus object an entry is made of: a stream, a key-value bucket (stream KV_<name>) and an object store
|
||||
// (stream OBJ_<name>).
|
||||
const (
|
||||
KindStream = "stream"
|
||||
KindBucket = "key-value"
|
||||
KindObjects = "object-store"
|
||||
)
|
||||
|
||||
// Separator joins a module's name and a local name into a bus name (design 25 § 12.1: `<module>_<local>`); a local
|
||||
// name holding it is refused, so no two modules' names meet.
|
||||
const Separator = "_"
|
||||
|
||||
// Fixed settings of the templates (design 25 § 12.1's table): never options.
|
||||
const (
|
||||
MiB = int64(1) << 20
|
||||
StateCap = 64 * MiB
|
||||
ValueMost = 256 << 10
|
||||
EntryMost = 256 << 10
|
||||
DefaultCapMiB = 1024
|
||||
EventsDays = 7
|
||||
)
|
||||
|
||||
// Settings are one bus object's settings, as asked and as held: what check compares.
|
||||
type Settings struct {
|
||||
// MaxBytes is its cap in bytes; MaxAgeSeconds its age, 0 for none.
|
||||
MaxBytes int64 `json:"max-bytes"`
|
||||
MaxAgeSeconds int64 `json:"max-age-seconds,omitempty"`
|
||||
// History is how many values a key keeps (a bucket's), or how many messages a subject keeps (a snapshot's); 0 none.
|
||||
History int64 `json:"history,omitempty"`
|
||||
// Discard is what it does when full: "new" (refuses) or "old" (drops the oldest).
|
||||
Discard string `json:"discard"`
|
||||
// WorkQueue says a message is removed once taken.
|
||||
WorkQueue bool `json:"work-queue,omitempty"`
|
||||
// MaxMsgSize is the largest message (a value, a log entry), 0 none.
|
||||
MaxMsgSize int64 `json:"max-msg-size,omitempty"`
|
||||
// DuplicatesSeconds is a work queue's duplicate window.
|
||||
DuplicatesSeconds int64 `json:"duplicates-seconds,omitempty"`
|
||||
// KeyTTL says a bucket's key may be given a lifetime of its own.
|
||||
KeyTTL bool `json:"key-ttl,omitempty"`
|
||||
}
|
||||
|
||||
// The words of Discard.
|
||||
const (
|
||||
DiscardNew = "new"
|
||||
DiscardOld = "old"
|
||||
)
|
||||
|
||||
// Refuses says the object refuses a new message when full: the ones the fill pass reads (ADR 0595 §6).
|
||||
func (s Settings) Refuses() bool { return s.Discard == DiscardNew }
|
||||
|
||||
// Object is one bus object an entry is made of: its kind, its name as the kind names it (a bucket by its bucket name,
|
||||
// a stream by its stream name), its subjects for a stream, and its settings.
|
||||
type Object struct {
|
||||
Kind string `json:"kind"`
|
||||
Name string `json:"name"`
|
||||
Subjects []string `json:"subjects,omitempty"`
|
||||
Settings Settings `json:"settings"`
|
||||
}
|
||||
|
||||
// Stream is the stream the bus keeps the object on: KV_<bucket>, OBJ_<store>, or the stream itself.
|
||||
func (o Object) Stream() string {
|
||||
switch o.Kind {
|
||||
case KindBucket:
|
||||
return "KV_" + o.Name
|
||||
case KindObjects:
|
||||
return "OBJ_" + o.Name
|
||||
}
|
||||
return o.Name
|
||||
}
|
||||
|
||||
// Plan is what one entry is on the bus: its binding (the names the module receives) and the objects it is made of.
|
||||
type Plan struct {
|
||||
Module string `json:"module"`
|
||||
Local string `json:"local"`
|
||||
Storage string `json:"storage"`
|
||||
Template string `json:"template,omitempty"`
|
||||
Binding membership.BusBinding `json:"binding"`
|
||||
Objects []Object `json:"objects"`
|
||||
Entry catalogue.BusStateEntry `json:"entry"`
|
||||
}
|
||||
|
||||
// BusName is `<module>_<local>`, the name the mesh gives an entry on the bus (design 25 § 12.1).
|
||||
func BusName(module, local string) string { return module + Separator + local }
|
||||
|
||||
// StreamSubject is the subject a stream entry's messages are under (`mesh.bus.<module>.<local>`, messages on its `.>`).
|
||||
func StreamSubject(module, local string) string { return "mesh.bus." + module + "." + local }
|
||||
|
||||
// PlanOf is the plan of one entry of module, or why it cannot be made: a local name holding the separator, a storage
|
||||
// type and template that do not go together, an option of the wrong shape. The judge refuses each of these first
|
||||
// (ADR 0595 §8); this is the keeper's own guard, so a manifest judged by an older judge is refused, never half-made.
|
||||
func PlanOf(module string, e catalogue.BusStateEntry) (Plan, error) {
|
||||
if e.Name == "" || strings.Contains(e.Name, Separator) || strings.ContainsAny(e.Name, ".*> ") {
|
||||
return Plan{}, fmt.Errorf("the local name %q is empty or holds %q, a dot, a wildcard or a space: the mesh names it %s on the bus, and two modules' names must never meet", e.Name, Separator, BusName(module, "<local>"))
|
||||
}
|
||||
name := BusName(module, e.Name)
|
||||
p := Plan{Module: module, Local: e.Name, Storage: e.Storage, Template: e.Template, Entry: e,
|
||||
Binding: membership.BusBinding{Local: e.Name, Storage: e.Storage, Template: e.Template}}
|
||||
opts := options{raw: e.Options}
|
||||
conf := options{raw: e.Configuration}
|
||||
streamOf := func(s Settings) Object {
|
||||
subject := StreamSubject(module, e.Name)
|
||||
p.Binding.Stream, p.Binding.Subject = name, subject
|
||||
return Object{Kind: KindStream, Name: name, Subjects: []string{subject + ".>"}, Settings: s}
|
||||
}
|
||||
switch {
|
||||
case e.Storage == catalogue.StorageKeyValue && e.Template == catalogue.TemplateState:
|
||||
s, err := stateSettings(opts)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
p.Binding.Bucket = name
|
||||
p.Objects = []Object{{Kind: KindBucket, Name: name, Settings: s}}
|
||||
case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateEvents:
|
||||
age, err := opts.int(catalogue.OptionAgeDays, EventsDays)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
capMiB, err := opts.int(catalogue.OptionCapMiB, DefaultCapMiB)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
p.Objects = []Object{streamOf(Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: DiscardOld})}
|
||||
case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateWorkQueue:
|
||||
capMiB, err := opts.int(catalogue.OptionCapMiB, DefaultCapMiB)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
age, err := opts.int(catalogue.OptionAgeDays, 0)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
dup, err := opts.int(catalogue.OptionDuplicates, 0)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
p.Objects = []Object{streamOf(Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: DiscardNew,
|
||||
WorkQueue: true, DuplicatesSeconds: dup})}
|
||||
case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateLog:
|
||||
s, err := logSettings(opts)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
p.Objects = []Object{streamOf(s)}
|
||||
case e.Storage == catalogue.StorageObjectStore && e.Template == catalogue.TemplateObjects:
|
||||
capMiB, err := opts.int(catalogue.OptionCapMiB, 0)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
if capMiB <= 0 {
|
||||
return Plan{}, fmt.Errorf("objects %q gives no %s, which the template objects requires", e.Name, catalogue.OptionCapMiB)
|
||||
}
|
||||
p.Binding.ObjectStore = name
|
||||
p.Objects = []Object{{Kind: KindObjects, Name: name, Settings: Settings{MaxBytes: capMiB * MiB, Discard: DiscardNew}}}
|
||||
case e.Storage == catalogue.StorageLedger && e.Template == catalogue.TemplateLedger:
|
||||
history, err := logSettings(opts)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
// Keeping days only in a ledger's own configuration (design 25 § 12.1).
|
||||
age, err := conf.int(catalogue.OptionAgeDays, 0)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
history.MaxAgeSeconds = days(age)
|
||||
state, err := stateSettings(options{})
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
subject := buscompose.LedgerHistorySubject(module, e.Name)
|
||||
snapshot := buscompose.LedgerSnapshotSubject(module, e.Name)
|
||||
p.Binding.History, p.Binding.Subject = name, subject
|
||||
p.Binding.State = name + "_state"
|
||||
p.Binding.Snapshot = snapshot
|
||||
p.Binding.Types = e.Types
|
||||
p.Objects = []Object{
|
||||
{Kind: KindStream, Name: name, Subjects: []string{subject + ".>"}, Settings: history},
|
||||
{Kind: KindBucket, Name: p.Binding.State, Settings: state},
|
||||
}
|
||||
var indexes []string
|
||||
for _, t := range e.Types {
|
||||
for _, ix := range t.Indexes {
|
||||
indexes = append(indexes, ix.Name)
|
||||
}
|
||||
}
|
||||
sort.Strings(indexes)
|
||||
for _, ix := range indexes {
|
||||
if p.Binding.Indexes == nil {
|
||||
p.Binding.Indexes = map[string]string{}
|
||||
}
|
||||
if _, twice := p.Binding.Indexes[ix]; twice {
|
||||
continue
|
||||
}
|
||||
p.Binding.Indexes[ix] = name + "_index-" + ix
|
||||
p.Objects = append(p.Objects, Object{Kind: KindBucket, Name: p.Binding.Indexes[ix], Settings: state})
|
||||
}
|
||||
// The snapshot: the latest state per instance, never expiring (ADR 0597 §8), one message per subject.
|
||||
p.Objects = append(p.Objects, Object{Kind: KindStream, Name: name + "_snapshot", Subjects: []string{snapshot + ".>"},
|
||||
Settings: Settings{MaxBytes: history.MaxBytes, History: 1, Discard: DiscardOld}})
|
||||
case e.Storage == catalogue.StorageStream && e.Template == "":
|
||||
// A bare stream: its own configuration, which must give when full, a cap and an age (design 25 § 12.1).
|
||||
var full string
|
||||
if raw, ok := conf.raw[catalogue.ConfigWhenFull]; ok {
|
||||
_ = json.Unmarshal(raw, &full)
|
||||
}
|
||||
capMiB, err := conf.int(catalogue.OptionCapMiB, 0)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
age, err := conf.int(catalogue.OptionAgeDays, -1)
|
||||
if err != nil {
|
||||
return Plan{}, err
|
||||
}
|
||||
discard := map[string]string{catalogue.WhenFullRefuse: DiscardNew, catalogue.WhenFullDropOldest: DiscardOld}[full]
|
||||
if discard == "" || capMiB <= 0 || age < 0 {
|
||||
return Plan{}, fmt.Errorf("the bare stream %q must give %s (refuse or drop-oldest), %s and %s in its configuration", e.Name,
|
||||
catalogue.ConfigWhenFull, catalogue.OptionCapMiB, catalogue.OptionAgeDays)
|
||||
}
|
||||
p.Objects = []Object{streamOf(Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: discard})}
|
||||
default:
|
||||
return Plan{}, fmt.Errorf("the storage type %q with the template %q is none the bus module makes (design 25 § 12.1)", e.Storage, e.Template)
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
func stateSettings(o options) (Settings, error) {
|
||||
history, err := o.int(catalogue.OptionPastValues, 1)
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
ttl, err := o.int(catalogue.OptionTimeToLive, 0)
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
keyTTL, err := o.bool(catalogue.OptionLifetimePerKey)
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
if perMachine, err := o.bool(catalogue.OptionPerMachine); err != nil || perMachine {
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
return Settings{}, fmt.Errorf("the option %s (one bucket per machine) is not made by the bus module yet: the module keeps one bucket", catalogue.OptionPerMachine)
|
||||
}
|
||||
return Settings{MaxBytes: StateCap, MaxAgeSeconds: ttl, History: history, Discard: DiscardNew, MaxMsgSize: ValueMost, KeyTTL: keyTTL}, nil
|
||||
}
|
||||
|
||||
func logSettings(o options) (Settings, error) {
|
||||
capMiB, err := o.int(catalogue.OptionCapMiB, DefaultCapMiB)
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
age, err := o.int(catalogue.OptionAgeDays, 0)
|
||||
if err != nil {
|
||||
return Settings{}, err
|
||||
}
|
||||
return Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: DiscardNew, MaxMsgSize: EntryMost}, nil
|
||||
}
|
||||
|
||||
func days(n int64) int64 {
|
||||
if n <= 0 {
|
||||
return 0
|
||||
}
|
||||
return int64((time.Duration(n) * 24 * time.Hour) / time.Second)
|
||||
}
|
||||
|
||||
// options reads a template's options as the manifest wrote them.
|
||||
type options struct{ raw map[string]json.RawMessage }
|
||||
|
||||
func (o options) int(name string, fallback int64) (int64, error) {
|
||||
raw, ok := o.raw[name]
|
||||
if !ok {
|
||||
return fallback, nil
|
||||
}
|
||||
n, err := strconv.ParseInt(strings.TrimSpace(string(raw)), 10, 64)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("the option %s is %s, not a whole number", name, raw)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (o options) bool(name string) (bool, error) {
|
||||
raw, ok := o.raw[name]
|
||||
if !ok {
|
||||
return false, nil
|
||||
}
|
||||
var b bool
|
||||
if err := json.Unmarshal(raw, &b); err != nil {
|
||||
return false, fmt.Errorf("the option %s is %s, not true or false", name, raw)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
// Grows is whether held reaches asked by growing alone (a larger cap, a longer age, more past values), so the keeper
|
||||
// applies it without a person (ADR 0595 §6); false for any other difference, which is a drift a person decides. Equal
|
||||
// settings grow by nothing and are no difference (Differs).
|
||||
func Grows(asked, held Settings) bool {
|
||||
if asked.Discard != held.Discard || asked.WorkQueue != held.WorkQueue || asked.MaxMsgSize != held.MaxMsgSize ||
|
||||
asked.DuplicatesSeconds != held.DuplicatesSeconds || asked.KeyTTL != held.KeyTTL {
|
||||
return false
|
||||
}
|
||||
return asked.MaxBytes >= held.MaxBytes && longer(asked.MaxAgeSeconds, held.MaxAgeSeconds) && asked.History >= held.History
|
||||
}
|
||||
|
||||
// longer is whether age a keeps at least as long as b, 0 being for ever.
|
||||
func longer(a, b int64) bool {
|
||||
if a == 0 {
|
||||
return true
|
||||
}
|
||||
return b != 0 && a >= b
|
||||
}
|
||||
|
||||
// Differs says how held differs from asked, field by field, or "" when it does not.
|
||||
func Differs(asked, held Settings) string {
|
||||
var out []string
|
||||
add := func(name string, a, h any) {
|
||||
if fmt.Sprint(a) != fmt.Sprint(h) {
|
||||
out = append(out, fmt.Sprintf("%s asked %v, held %v", name, a, h))
|
||||
}
|
||||
}
|
||||
add("max-bytes", asked.MaxBytes, held.MaxBytes)
|
||||
add("max-age-seconds", asked.MaxAgeSeconds, held.MaxAgeSeconds)
|
||||
add("history", asked.History, held.History)
|
||||
add("discard", asked.Discard, held.Discard)
|
||||
add("work-queue", asked.WorkQueue, held.WorkQueue)
|
||||
add("max-msg-size", asked.MaxMsgSize, held.MaxMsgSize)
|
||||
add("duplicates-seconds", asked.DuplicatesSeconds, held.DuplicatesSeconds)
|
||||
add("key-ttl", asked.KeyTTL, held.KeyTTL)
|
||||
return strings.Join(out, "; ")
|
||||
}
|
||||
Reference in New Issue
Block a user