From ff4d042be537c2ebb17d28c1d63a9ee423089786 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 11:49:31 +0100 Subject: [PATCH] CAT-N3: the keeper creates each template's bus objects under the module's names and says created or refused MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each bus-state entry is planned into its objects (design 25 § 12.1): state a bucket _; events, work queue, log and a bare stream a stream _ on mesh.bus...>; 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 --- modules/nats/internal/keeper/jetstream.go | 159 ++++++ modules/nats/internal/keeper/keeper.go | 599 ++++++++++++++++++++ modules/nats/internal/keeper/keeper_test.go | 343 +++++++++++ modules/nats/internal/keeper/plan.go | 349 ++++++++++++ 4 files changed, 1450 insertions(+) create mode 100644 modules/nats/internal/keeper/jetstream.go create mode 100644 modules/nats/internal/keeper/keeper.go create mode 100644 modules/nats/internal/keeper/keeper_test.go create mode 100644 modules/nats/internal/keeper/plan.go diff --git a/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go new file mode 100644 index 00000000..9f13e6d7 --- /dev/null +++ b/modules/nats/internal/keeper/jetstream.go @@ -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 +} diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go new file mode 100644 index 00000000..14efcae5 --- /dev/null +++ b/modules/nats/internal/keeper/keeper.go @@ -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, ` : `: 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.._`. +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 +} diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go new file mode 100644 index 00000000..e4743533 --- /dev/null +++ b/modules/nats/internal/keeper/keeper_test.go @@ -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 `_`, 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) + }) + } +} diff --git a/modules/nats/internal/keeper/plan.go b/modules/nats/internal/keeper/plan.go new file mode 100644 index 00000000..536ca32b --- /dev/null +++ b/modules/nats/internal/keeper/plan.go @@ -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_) and an object store +// (stream OBJ_). +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: `_`); 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_, OBJ_, 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 `_`, 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..`, 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, "")) + } + 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, "; ") +}