Merge pull request 'Let a module keep a log the controller creates and never removes, and warn before a bucket or log is full (ADR 0297)' (#231) from adr-0297-module-logs into main
This commit was merged in pull request #231.
This commit is contained in:
@@ -0,0 +1,207 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
)
|
||||
|
||||
// **A filling bucket or log is said before it is full** (novox/hq ADR 0297 §6, issue 501).
|
||||
//
|
||||
// A module's bucket and its log each have a cap on the bus, and when either is full the bus refuses
|
||||
// the next write: the module stops doing what it writes for, and nothing said so before. On
|
||||
// 2026-10-11 the issue tracker's bucket held a sixth of its cap after two days, growing toward a stop
|
||||
// nobody would have heard coming. So the self-check reads every module's bucket and log, and one at
|
||||
// fillRaiseAt of its cap or more raises a condition naming the module, the bucket or log, and how full
|
||||
// it is.
|
||||
//
|
||||
// **Cleared by observation below fillClearBelow, not below fillRaiseAt**: a bucket or log hovering at
|
||||
// its threshold would otherwise be raised and cleared on every run. Between the two, an open condition
|
||||
// is kept and none is raised.
|
||||
//
|
||||
// **One that cannot be read is said, and the rest are still judged.** An unanswered question about one
|
||||
// stream is no reason to know nothing of the others: it is a finding of its own, naming the bucket or
|
||||
// log and why, and a fill condition already open for it is kept, because not knowing is not a pass.
|
||||
|
||||
// The fill thresholds, in percent of a bucket's or log's cap.
|
||||
const (
|
||||
fillRaiseAt = 75
|
||||
fillClearBelow = 70
|
||||
)
|
||||
|
||||
// The conditions the fill probe raises: one filling, and one that could not be read.
|
||||
const (
|
||||
kindBucketOrLogFilling = "bucket-or-log-filling"
|
||||
kindBucketOrLogUnread = "bucket-or-log-unread"
|
||||
)
|
||||
|
||||
// probeFillID is the probe's id in the registry.
|
||||
const probeFillID = "D-fill"
|
||||
|
||||
// bucketOrLog is one module's bucket or log as the fill probe reads it.
|
||||
type bucketOrLog struct {
|
||||
Module string
|
||||
// Kind is `bucket` or `log`; Name its local name; Stream the stream it is on the bus.
|
||||
Kind, Name, Stream string
|
||||
}
|
||||
|
||||
// filled is how full one bucket or log is, read from the bus.
|
||||
type filled struct {
|
||||
Bytes, Max uint64
|
||||
}
|
||||
|
||||
// percent is how full, in whole percent, rounded down.
|
||||
func (f filled) percent() uint64 {
|
||||
if f.Max == 0 {
|
||||
return 0
|
||||
}
|
||||
return f.Bytes * 100 / f.Max
|
||||
}
|
||||
|
||||
// fillKey is where a bucket's or log's fill condition is kept.
|
||||
func fillKey(s bucketOrLog) string { return conditions.Key(conditions.ScopeBus, s.Stream, "filling") }
|
||||
|
||||
// fillObservation is what one bucket's or log's fill says: a finding at fillRaiseAt or above, and,
|
||||
// while its condition is open, at fillClearBelow or above too; nothing otherwise, which is what clears
|
||||
// it. The operator's words are the kind's (plain_words.go); the summary and the evidence name the
|
||||
// module, the bucket or log, and the fill.
|
||||
func fillObservation(s bucketOrLog, f filled, open bool) (conditions.Observation, bool) {
|
||||
if f.Max == 0 {
|
||||
return conditions.Observation{}, false // one with no cap cannot fill
|
||||
}
|
||||
// Compared in bytes, not rounded percent: 74.9% is not 75%.
|
||||
atRaise := f.Bytes*100 >= f.Max*fillRaiseAt
|
||||
aboveClear := f.Bytes*100 >= f.Max*fillClearBelow
|
||||
if !atRaise && !(open && aboveClear) {
|
||||
return conditions.Observation{}, false
|
||||
}
|
||||
pct := f.percent()
|
||||
return conditions.Observation{
|
||||
Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling,
|
||||
Severity: conditions.Warning,
|
||||
Summary: fmt.Sprintf("%s's %s %s holds %s of %s, %d%%: at its cap the bus refuses the module's next write",
|
||||
s.Module, s.Kind, s.Name, mibWords(f.Bytes), mibWords(f.Max), pct),
|
||||
Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", s.Module, s.Kind, s.Name,
|
||||
s.Stream, f.Bytes, f.Max, pct),
|
||||
}, true
|
||||
}
|
||||
|
||||
// unreadObservation says one bucket or log could not be read, and why. Asked over the network, so one
|
||||
// look can be wrong: raised on the second look in a row (confirm.go).
|
||||
func unreadObservation(s bucketOrLog, why error) conditions.Observation {
|
||||
return conditions.Observation{
|
||||
Scope: conditions.ScopeBus, ID: s.Stream, Token: "unread", Kind: kindBucketOrLogUnread,
|
||||
Severity: conditions.Warning, Confirm: true,
|
||||
Summary: fmt.Sprintf("%s's %s %s could not be read, so how full it is is not known", s.Module, s.Kind, s.Name),
|
||||
Said: fmt.Sprintf("module %s, %s %s (stream %s): %v", s.Module, s.Kind, s.Name, s.Stream, why),
|
||||
}
|
||||
}
|
||||
|
||||
// keptFilling is an open fill condition said again for a bucket or log that could not be read this
|
||||
// time: not knowing how full it is now is no reason to say it has room. Only ever made from a condition
|
||||
// read back as open — its own summary and severity, never words made up for it.
|
||||
func keptFilling(s bucketOrLog, open conditions.Condition, why error) conditions.Observation {
|
||||
return conditions.Observation{
|
||||
Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindBucketOrLogFilling,
|
||||
Severity: open.Severity, Summary: open.Summary,
|
||||
Said: fmt.Sprintf("module %s, %s %s (stream %s): not read this time (%v); kept open as it was",
|
||||
s.Module, s.Kind, s.Name, s.Stream, why),
|
||||
}
|
||||
}
|
||||
|
||||
// mibWords is a size as a person reads it, in MiB with one decimal.
|
||||
func mibWords(b uint64) string {
|
||||
return fmt.Sprintf("%.1f MiB", float64(b)/(1024*1024))
|
||||
}
|
||||
|
||||
// readFill is how full one bucket or log is on the bus; found false when it is not on the bus.
|
||||
type readFill func(ctx context.Context, s bucketOrLog) (f filled, found bool, err error)
|
||||
|
||||
// openFill is the fill condition open under a key, if one is, or why it could not be read.
|
||||
type openFill func(ctx context.Context, key string) (conditions.Condition, bool, error)
|
||||
|
||||
// judgeFills judges every bucket and log on its own: one that cannot be read is said as unread, with
|
||||
// its fill condition kept only when that condition was read back and is open — when it cannot be read
|
||||
// either, nothing is said of its fill, which the unread finding already covers. Every other is judged
|
||||
// by its fill; for one of those, a condition that cannot be read is taken as open, so an unknown never
|
||||
// clears a fill measured between fillClearBelow and fillRaiseAt.
|
||||
func judgeFills(ctx context.Context, all []bucketOrLog, read readFill, open openFill) []conditions.Observation {
|
||||
var out []conditions.Observation
|
||||
for _, s := range all {
|
||||
f, found, err := read(ctx, s)
|
||||
if err != nil {
|
||||
out = append(out, unreadObservation(s, err))
|
||||
if c, isOpen, cerr := open(ctx, fillKey(s)); cerr == nil && isOpen {
|
||||
out = append(out, keptFilling(s, c, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !found {
|
||||
continue // not on the bus: created on the controller's next raise, and nothing there can fill
|
||||
}
|
||||
_, isOpen, cerr := open(ctx, fillKey(s))
|
||||
isOpen = isOpen || cerr != nil
|
||||
if o, said := fillObservation(s, f, isOpen); said {
|
||||
out = append(out, o)
|
||||
}
|
||||
}
|
||||
sort.SliceStable(out, func(i, j int) bool { return out[i].Key() < out[j].Key() })
|
||||
return out
|
||||
}
|
||||
|
||||
// declaredBucketsAndLogs is every module's bucket and log the catalogue declares, by its stream.
|
||||
func declaredBucketsAndLogs(ctx context.Context, d *doctor) ([]bucketOrLog, error) {
|
||||
buckets, err := d.open.inventory.DeclaredBuckets(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
logs, err := d.open.inventory.DeclaredLogs(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []bucketOrLog
|
||||
for _, b := range buckets {
|
||||
out = append(out, bucketOrLog{Module: b.Module, Kind: "bucket", Name: b.Name, Stream: "KV_" + b.Bucket()})
|
||||
}
|
||||
for _, l := range logs {
|
||||
out = append(out, bucketOrLog{Module: l.Module, Kind: "log", Name: l.Name, Stream: l.Stream()})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// probeFill reads every module's bucket and log on the bus and says each one filling toward its cap,
|
||||
// and each one that could not be read.
|
||||
func probeFill(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
|
||||
all, err := declaredBucketsAndLogs(ctx, d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
js := d.js.Context()
|
||||
read := func(ctx context.Context, s bucketOrLog) (filled, bool, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return filled{}, false, err
|
||||
}
|
||||
info, err := js.StreamInfo(s.Stream, nats.Context(ctx))
|
||||
switch {
|
||||
case errors.Is(err, nats.ErrStreamNotFound):
|
||||
return filled{}, false, nil
|
||||
case err != nil:
|
||||
return filled{}, false, err
|
||||
case info.Config.MaxBytes <= 0:
|
||||
return filled{}, true, nil
|
||||
}
|
||||
return filled{Bytes: info.State.Bytes, Max: uint64(info.Config.MaxBytes)}, true, nil
|
||||
}
|
||||
open := func(ctx context.Context, key string) (conditions.Condition, bool, error) {
|
||||
if d.keeper == nil {
|
||||
return conditions.Condition{}, false, nil
|
||||
}
|
||||
return d.keeper.Get(ctx, key)
|
||||
}
|
||||
return judgeFills(ctx, all, read, open), nil
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
)
|
||||
|
||||
// **A filling log or bucket is said at three quarters of its cap and cleared below seven tenths**
|
||||
// (novox/hq ADR 0297 §6): raised at 75% naming the module, the bucket or log and its fill; not at 74.9%;
|
||||
// kept between 70% and 75% while it is open, and not raised there when it is not; and cleared — said no
|
||||
// more — once it reads below 70%, open or not.
|
||||
func TestAFillingBucketOrLogIsRaisedAtThreeQuartersAndClearedBelowSevenTenths(t *testing.T) {
|
||||
const mib = 1024 * 1024
|
||||
log := bucketOrLog{Module: "mesh-issues", Kind: "log", Name: "changes", Stream: "LOG_mesh-issues_changes"}
|
||||
bucket := bucketOrLog{Module: "mesh-issues", Kind: "bucket", Name: "issues", Stream: "KV_mesh-issues_issues"}
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
of bucketOrLog
|
||||
bytes uint64
|
||||
max uint64
|
||||
open bool
|
||||
said bool
|
||||
}{
|
||||
{"a log at exactly 75%", log, 768 * mib, 1024 * mib, false, true},
|
||||
{"a bucket at exactly 75%", bucket, 48 * mib, 64 * mib, false, true},
|
||||
{"a bucket full", bucket, 64 * mib, 64 * mib, false, true},
|
||||
{"a log just under 75%", log, 768*mib - 1, 1024 * mib, false, false},
|
||||
{"a bucket at 72%, not open", bucket, 64 * mib * 72 / 100, 64 * mib, false, false},
|
||||
{"a bucket at 72%, open: kept", bucket, 64 * mib * 72 / 100, 64 * mib, true, true},
|
||||
{"a log at exactly 70%, open: kept", log, 700 * mib, 1000 * mib, true, true},
|
||||
{"a log just under 70%, open: cleared", log, 700*mib - 1, 1000 * mib, true, false},
|
||||
{"a bucket at 10%, open: cleared", bucket, 64 * mib / 10, 64 * mib, true, false},
|
||||
{"a bucket with no cap", bucket, 100, 0, true, false},
|
||||
} {
|
||||
o, said := fillObservation(c.of, filled{Bytes: c.bytes, Max: c.max}, c.open)
|
||||
if said != c.said {
|
||||
t.Errorf("%s: said %v, want %v", c.name, said, c.said)
|
||||
continue
|
||||
}
|
||||
if !said {
|
||||
continue
|
||||
}
|
||||
if o.Key() != fillKey(c.of) || o.Kind != kindBucketOrLogFilling || o.Severity != conditions.Warning {
|
||||
t.Errorf("%s: raised as %s (%s, %s)", c.name, o.Key(), o.Kind, o.Severity)
|
||||
}
|
||||
for _, part := range []string{c.of.Module, c.of.Kind + " " + c.of.Name, "%", " of "} {
|
||||
if !strings.Contains(o.Summary, part) {
|
||||
t.Errorf("%s: does not name %q: %q", c.name, part, o.Summary)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(o.Said, "bytes") {
|
||||
t.Errorf("%s: the evidence does not say the fill in bytes: %q", c.name, o.Said)
|
||||
}
|
||||
}
|
||||
o, _ := fillObservation(log, filled{Bytes: 800 * mib, Max: 1024 * mib}, false)
|
||||
if !strings.Contains(o.Summary, "800.0 MiB of 1024.0 MiB, 78%") {
|
||||
t.Errorf("the fill is said as %q", o.Summary)
|
||||
}
|
||||
}
|
||||
|
||||
// **One bucket or log that cannot be read does not stop the others being judged** (review of #231): it is
|
||||
// said as unread with its reason, a fill condition open for it is kept, and every other bucket and log
|
||||
// is judged by its own fill.
|
||||
func TestAnUnreadableBucketOrLogIsSaidAndTheRestAreStillJudged(t *testing.T) {
|
||||
const mib = 1024 * 1024
|
||||
broken := bucketOrLog{Module: "a", Kind: "bucket", Name: "broken", Stream: "KV_a_broken"}
|
||||
brokenOpen := bucketOrLog{Module: "a", Kind: "log", Name: "was-filling", Stream: "LOG_a_was-filling"}
|
||||
full := bucketOrLog{Module: "b", Kind: "log", Name: "changes", Stream: "LOG_b_changes"}
|
||||
empty := bucketOrLog{Module: "c", Kind: "bucket", Name: "quiet", Stream: "KV_c_quiet"}
|
||||
absent := bucketOrLog{Module: "d", Kind: "bucket", Name: "not-yet", Stream: "KV_d_not-yet"}
|
||||
refusal := errors.New("the bus did not answer")
|
||||
read := func(_ context.Context, s bucketOrLog) (filled, bool, error) {
|
||||
switch s {
|
||||
case broken, brokenOpen:
|
||||
return filled{}, false, refusal
|
||||
case full:
|
||||
return filled{Bytes: 900 * mib, Max: 1000 * mib}, true, nil
|
||||
case empty:
|
||||
return filled{Bytes: 1, Max: 64 * mib}, true, nil
|
||||
}
|
||||
return filled{}, false, nil
|
||||
}
|
||||
open := func(_ context.Context, key string) (conditions.Condition, bool, error) {
|
||||
if key == fillKey(brokenOpen) {
|
||||
return conditions.Condition{Key: key, Severity: conditions.Warning,
|
||||
Summary: "a's log was-filling holds 800.0 MiB of 1024.0 MiB, 78%"}, true, nil
|
||||
}
|
||||
return conditions.Condition{}, false, nil
|
||||
}
|
||||
got := map[string]conditions.Observation{}
|
||||
for _, o := range judgeFills(context.Background(), []bucketOrLog{broken, brokenOpen, full, empty, absent}, read, open) {
|
||||
got[o.Key()] = o
|
||||
}
|
||||
for _, s := range []bucketOrLog{broken, brokenOpen} {
|
||||
o, said := got[conditions.Key(conditions.ScopeBus, s.Stream, "unread")]
|
||||
if !said || o.Kind != kindBucketOrLogUnread || !o.Confirm || !strings.Contains(o.Said, refusal.Error()) {
|
||||
t.Errorf("%s could not be read and was said as %+v", s.Stream, o)
|
||||
}
|
||||
}
|
||||
if o, kept := got[fillKey(brokenOpen)]; !kept || !strings.Contains(o.Said, "kept open") {
|
||||
t.Errorf("an open fill condition of a log not read this time was not kept: %+v", o)
|
||||
}
|
||||
if _, said := got[fillKey(broken)]; said {
|
||||
t.Error("a bucket not read and not filling was said filling")
|
||||
}
|
||||
if o, said := got[fillKey(full)]; !said || o.Kind != kindBucketOrLogFilling {
|
||||
t.Errorf("a log at 90%% beside an unreadable one was not judged: %+v", got)
|
||||
}
|
||||
if len(got) != 4 {
|
||||
t.Errorf("said %d findings, want 4 (two unread, one kept, one filling): %v", len(got), got)
|
||||
}
|
||||
}
|
||||
|
||||
// Both kinds have plain words of their own, which hold to the plain rule (novox/hq ADR 0253).
|
||||
func TestAFillingOrUnreadBucketOrLogIsSaidInPlainWords(t *testing.T) {
|
||||
for _, kind := range []string{kindBucketOrLogFilling, kindBucketOrLogUnread} {
|
||||
wording, has := plainWordings[kind]
|
||||
if !has {
|
||||
t.Errorf("%s has no plain words", kind)
|
||||
continue
|
||||
}
|
||||
if why, ok := conditions.PlainWords(wording(conditions.Observation{Kind: kind})); !ok {
|
||||
t.Errorf("%s: %s", kind, why)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **When neither the bucket or log nor its condition can be read, nothing is said of its fill** (second
|
||||
// review of #231): the unread finding covers it, and no fill is invented for it. For one that is read and
|
||||
// measured between seven tenths and three quarters, a condition that cannot be read is taken as open, so
|
||||
// an unknown never clears it.
|
||||
func TestNothingIsSaidOfAFillWhenNeitherItNorItsConditionCanBeRead(t *testing.T) {
|
||||
const mib = 1024 * 1024
|
||||
broken := bucketOrLog{Module: "a", Kind: "log", Name: "changes", Stream: "LOG_a_changes"}
|
||||
between := bucketOrLog{Module: "b", Kind: "bucket", Name: "issues", Stream: "KV_b_issues"}
|
||||
read := func(_ context.Context, s bucketOrLog) (filled, bool, error) {
|
||||
if s == broken {
|
||||
return filled{}, false, errors.New("the bus did not answer")
|
||||
}
|
||||
return filled{Bytes: 72 * mib, Max: 100 * mib}, true, nil
|
||||
}
|
||||
open := func(context.Context, string) (conditions.Condition, bool, error) {
|
||||
return conditions.Condition{}, false, errors.New("the conditions bucket did not answer")
|
||||
}
|
||||
got := map[string]conditions.Observation{}
|
||||
for _, o := range judgeFills(context.Background(), []bucketOrLog{broken, between}, read, open) {
|
||||
got[o.Key()] = o
|
||||
}
|
||||
if o, said := got[fillKey(broken)]; said {
|
||||
t.Fatalf("a fill was said for a log whose fill and condition could not be read: %+v", o)
|
||||
}
|
||||
if _, said := got[conditions.Key(conditions.ScopeBus, broken.Stream, "unread")]; !said {
|
||||
t.Error("the log that could not be read was not said as unread")
|
||||
}
|
||||
if _, kept := got[fillKey(between)]; !kept {
|
||||
t.Error("a bucket at 72% whose condition could not be read was cleared")
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Errorf("said %d findings, want 2: %v", len(got), got)
|
||||
}
|
||||
}
|
||||
@@ -179,6 +179,14 @@ func moduleCheckFor(paths []string, longestMachine int, out io.Writer) error {
|
||||
}
|
||||
fmt.Fprintf(out, ", keeps state %s", strings.Join(kept, ", "))
|
||||
}
|
||||
// And the logs it keeps, with their caps (novox/hq ADR 0297).
|
||||
if len(m.Logs) > 0 {
|
||||
kept := make([]string, 0, len(m.Logs))
|
||||
for _, l := range m.Logs {
|
||||
kept = append(kept, fmt.Sprintf("%s (%d MiB)", l.Name, l.Cap()))
|
||||
}
|
||||
fmt.Fprintf(out, ", keeps log %s", strings.Join(kept, ", "))
|
||||
}
|
||||
if len(m.Reads) > 0 {
|
||||
fmt.Fprintf(out, ", reads %s", strings.Join(m.Reads, ", "))
|
||||
}
|
||||
|
||||
@@ -140,6 +140,12 @@ var probeRegistry = []probe{
|
||||
{ID: "D-root", Asserts: "no agent can become root without a person on a machine where the router or a channel " +
|
||||
"proving its sender runs: not by its own account, and not through a tool that runs its command as an account " +
|
||||
"that can", From: "ADR 0259 §8", Kind: kindRootNotFree, Phase: 2, run: probeAgentRoot},
|
||||
// A module's bucket or log filling toward its cap (novox/hq ADR 0297 §6): said at three quarters, cleared
|
||||
// below seven tenths, so one at its threshold is not raised and cleared on every run. One that cannot be
|
||||
// read is said on its own, and every other is still judged.
|
||||
{ID: probeFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " +
|
||||
"said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501",
|
||||
Kind: kindBucketOrLogFilling, Raises: []string{kindBucketOrLogUnread}, Phase: 1, run: probeFill},
|
||||
{ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals",
|
||||
From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs},
|
||||
// The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is
|
||||
|
||||
@@ -614,6 +614,19 @@ var plainWordings = map[string]func(conditions.Observation) words{
|
||||
Explanation: "The bus refused messages from a part of the mesh, so what they carried did not happen.",
|
||||
Resolved: "Resolved: the bus takes the messages again"}
|
||||
}),
|
||||
kindBucketOrLogFilling: worded(func(o conditions.Observation) words {
|
||||
return words{Headline: "A module's bucket or log on the bus is filling up",
|
||||
Needs: "decide whether to raise its cap or have the module keep less.",
|
||||
Explanation: "A bucket or log a module keeps on the bus holds three quarters of its cap or more. When it " +
|
||||
"is full, the bus refuses what the module writes there.",
|
||||
Resolved: "It has room again"}
|
||||
}),
|
||||
kindBucketOrLogUnread: worded(func(o conditions.Observation) words {
|
||||
return words{Headline: "A module's bucket or log could not be read",
|
||||
Explanation: "The controller could not read how full a bucket or log a module keeps on the bus is, so it " +
|
||||
"cannot say whether it is filling up. It asks again at its next check.",
|
||||
Resolved: "It can be read again"}
|
||||
}),
|
||||
"stream-wrong": worded(func(o conditions.Observation) words {
|
||||
return words{Headline: "Part of the bus's storage is wrong",
|
||||
Needs: "check the machine the bus runs on; the details say what is missing.",
|
||||
|
||||
@@ -1311,6 +1311,15 @@ func issueMemberships(ctx context.Context, open *stores, server *link.Server, se
|
||||
if _, err := broker.RaiseBuckets(broker.OnConn(bus.Conn), buckets); err != nil {
|
||||
return fmt.Errorf("the modules' state could not be asserted on the bus: %w", err)
|
||||
}
|
||||
// **And every declared log, for the same reason** (novox/hq ADR 0297 §2): a membership names its
|
||||
// logs, and a module whose log does not exist fails its first append.
|
||||
logs, err := open.inventory.DeclaredLogs(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the modules' logs could not be read, so no log was asserted: %w", err)
|
||||
}
|
||||
if _, err := broker.RaiseLogs(broker.OnConn(bus.Conn), logs); err != nil {
|
||||
return fmt.Errorf("the modules' logs could not be asserted on the bus: %w", err)
|
||||
}
|
||||
// Every membership is tried, and the first failure named once.
|
||||
issued := 0
|
||||
refused := map[string]error{}
|
||||
@@ -1494,13 +1503,28 @@ func raiseTheBus(ctx context.Context, inv *inventory.Inventory, address string)
|
||||
fmt.Printf("the bus holds state nothing declares any more, kept because it is data: %s — "+
|
||||
"removing it is a person's act\n", strings.Join(undeclared, ", "))
|
||||
}
|
||||
// Every module's log (novox/hq ADR 0297), from the catalogue, as its buckets: one that nothing
|
||||
// declares any more is said and kept — a log is a module's record, and no path of the mesh removes it.
|
||||
logs, err := inv.DeclaredLogs(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
unlogged, err := broker.RaiseLogs(js, logs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(unlogged) > 0 {
|
||||
fmt.Printf("the bus holds logs nothing declares any more, kept because they are data: %s — "+
|
||||
"removing one is a person's act\n", strings.Join(unlogged, ", "))
|
||||
}
|
||||
// And how every module hears what it consumes: asserted with the rest above, counted here.
|
||||
hearing, err := moduleConsumerCount(ctx, inv)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Printf("the bus at %s has its streams, %d machine(s) can hear a declaration, %d module(s) "+
|
||||
"can hear what they consume, and %d bucket(s) of state\n", broker.BareAddress(address), len(names), hearing, len(buckets))
|
||||
"can hear what they consume, %d bucket(s) of state and %d log(s)\n", broker.BareAddress(address), len(names), hearing,
|
||||
len(buckets), len(logs))
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user