CAT-K12: the health on the ledger names the source and the move's last check, and fails for what the ledger owes
Unhealthy, naming the cause, for a ledger or the verdicts' stream that does not answer, a head behind its history, an index behind the ledger and a stream at its cap; the idle drafts, the parts waiting on the judge and the records the form-only check refused are counted and leave it healthy. A part waiting more than 30 minutes is asked of the operator once, cleared when it is judged. Issue 1966. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,210 @@
|
||||
package main
|
||||
|
||||
// The module's health once its records are on the ledger (novox/hq to-be 55 §1.1, ADR 0592 §33 changed by ADR 0597 §12):
|
||||
// unhealthy, naming the cause, when a kept thing does not answer, when the search index is behind the ledger, when a head lags
|
||||
// behind its history, and when the stream of the verdicts is at its cap. It names where the records are served from and the
|
||||
// result of the move's last check, and counts the drafts idle more than a week, the records the form-only check refused and the
|
||||
// parts waiting more than 30 minutes; such a part is asked of the operator once, and the ask clears when it is judged. A record
|
||||
// waiting for its review is not the module failing, so those counts leave it healthy.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||
)
|
||||
|
||||
// PartWaits is how long a part may wait on the judge before the operator is asked once.
|
||||
const PartWaits = 30 * time.Minute
|
||||
|
||||
// ledgerHealth adds what the ledger form owes to the verdict of the watch.
|
||||
func (k *KB) ledgerHealth(base HealthAnswer) HealthAnswer {
|
||||
ms, _ := k.Store.(*movableStore)
|
||||
l, err := k.ledger()
|
||||
if err != nil {
|
||||
if ms != nil {
|
||||
base.Why += "; the records are served from the " + SourceDatabase
|
||||
}
|
||||
return base
|
||||
}
|
||||
var broken, notes []string
|
||||
notes = append(notes, "the records are served from the ledger")
|
||||
// A kept thing that does not answer, and a head behind its history.
|
||||
var lag buscalls.Lagged
|
||||
if err := l.pipe.Call(buscalls.LedgerLag, buscalls.LedgerRequest{Ledger: LedgerName}, &lag); err != nil {
|
||||
broken = append(broken, "the ledger does not answer: "+err.Error())
|
||||
} else if len(lag.Instances) > 0 {
|
||||
names := make([]string, 0, 3)
|
||||
for i, in := range lag.Instances {
|
||||
if i == 3 {
|
||||
break
|
||||
}
|
||||
names = append(names, fmt.Sprintf("%s (state at %d of %d)", in.Instance, in.State, in.History))
|
||||
}
|
||||
broken = append(broken, fmt.Sprintf("%d head(s) are behind their history: %s", len(lag.Instances), strings.Join(names, ", ")))
|
||||
}
|
||||
// The index is behind the ledger when it holds fewer records than the ledger does.
|
||||
st := l.IndexState()
|
||||
l.mu.Lock()
|
||||
heads, indexed := len(l.heads), 0
|
||||
if l.index != nil {
|
||||
indexed = len(l.index.docsOf)
|
||||
}
|
||||
l.mu.Unlock()
|
||||
switch {
|
||||
case !st.Built:
|
||||
notes = append(notes, "the search index is being built")
|
||||
case heads > 0 && indexed < heads-reservedCount(l):
|
||||
broken = append(broken, fmt.Sprintf("the search index holds %d of %d records: it is behind the ledger", indexed, heads))
|
||||
default:
|
||||
notes = append(notes, fmt.Sprintf("the search index holds %d documents", st.Docs))
|
||||
}
|
||||
// The stream of the verdicts at its cap.
|
||||
var fill buscalls.StreamFilled
|
||||
if err := l.pipe.Call(buscalls.StreamFill, buscalls.StreamRequest{Stream: "reviews"}, &fill); err != nil {
|
||||
broken = append(broken, "the stream of the verdicts does not answer: "+err.Error())
|
||||
} else if fill.MaxBytes > 0 && fill.Bytes >= fill.MaxBytes {
|
||||
broken = append(broken, fmt.Sprintf("the stream of the verdicts is at its cap of %d bytes", fill.MaxBytes))
|
||||
}
|
||||
// The move's last check.
|
||||
if kv := k.States; kv != nil {
|
||||
if e, err := kv.State(findingsState).Get(checkKey); err == nil && e != nil {
|
||||
var c struct {
|
||||
Differences []string `json:"differences"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
if json.Unmarshal(e.Value, &c) == nil {
|
||||
if len(c.Differences) == 0 {
|
||||
notes = append(notes, "the last check of the database against the ledger was clean ("+c.At.Local().Format("2006-01-02 15:04")+")")
|
||||
} else {
|
||||
notes = append(notes, fmt.Sprintf("the last check of the database against the ledger found %d difference(s)", len(c.Differences)))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// What waits on a person, counted and left healthy.
|
||||
idle, waiting := k.waitingCounts()
|
||||
if idle > 0 {
|
||||
notes = append(notes, fmt.Sprintf("%d draft(s) untouched for more than a week", idle))
|
||||
}
|
||||
if waiting > 0 {
|
||||
notes = append(notes, fmt.Sprintf("%d part(s) waiting on the judge for more than 30 minutes", waiting))
|
||||
}
|
||||
if refused := k.formRefused(); refused > 0 {
|
||||
notes = append(notes, fmt.Sprintf("%d record(s) the form-only check refused", refused))
|
||||
}
|
||||
if len(broken) > 0 {
|
||||
base.State, base.Healthy = "unhealthy", false
|
||||
base.Why = strings.Join(broken, "; ") + "; " + strings.Join(notes, "; ")
|
||||
return base
|
||||
}
|
||||
base.Why += "; " + strings.Join(notes, "; ")
|
||||
return base
|
||||
}
|
||||
|
||||
func reservedCount(l *ledgerStore) int {
|
||||
n := 0
|
||||
for _, h := range l.heads {
|
||||
if st := h.Entry.Current.Status; st == StatusReserved || st == StatusWithdrawn || st == StatusRetracted {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// waitingCounts are the drafts idle more than DraftIdle and the parts waiting more than PartWaits; a part that waits is
|
||||
// asked of the operator once, and the ask is cleared when it is judged (the mark leaves `asked`).
|
||||
func (k *KB) waitingCounts() (idle, waiting int) {
|
||||
if k.States == nil {
|
||||
return 0, 0
|
||||
}
|
||||
now := k.now()
|
||||
if keys, err := k.States.State(draftsState).Keys(); err == nil {
|
||||
for _, key := range keys {
|
||||
if e, err := k.States.State(draftsState).Get(key); err == nil && e != nil {
|
||||
var d Draft
|
||||
if json.Unmarshal(e.Value, &d) == nil && now.Sub(d.Touched) > DraftIdle {
|
||||
idle++
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
live := map[string]bool{}
|
||||
if keys, err := k.States.State(askedState).Keys(); err == nil {
|
||||
for _, key := range keys {
|
||||
e, err := k.States.State(askedState).Get(key)
|
||||
if err != nil || e == nil {
|
||||
continue
|
||||
}
|
||||
var a struct {
|
||||
Part string `json:"part"`
|
||||
Record string `json:"record"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
if json.Unmarshal(e.Value, &a) != nil || now.Sub(a.At) <= PartWaits {
|
||||
continue
|
||||
}
|
||||
waiting++
|
||||
live[key] = true
|
||||
k.askPartWaits(key, a.Record, a.Part)
|
||||
}
|
||||
}
|
||||
k.clearPartWaits(live)
|
||||
return idle, waiting
|
||||
}
|
||||
|
||||
const partWaitPrefix = "part-wait."
|
||||
|
||||
// askPartWaits asks the operator once, through the operator channel, about a part waiting more than 30 minutes.
|
||||
func (k *KB) askPartWaits(key, record, part string) {
|
||||
if k.Notify == nil || k.Store == nil {
|
||||
return
|
||||
}
|
||||
ctx := context.Background()
|
||||
if first, err := k.Store.Told(ctx, partWaitPrefix+key); err != nil || !first {
|
||||
return
|
||||
}
|
||||
if err := k.Notify("knowledge."+partWaitPrefix+key, fmt.Sprintf("The review of %s waits on the judge: its part %q has waited more than 30 minutes.", record, part),
|
||||
"A review is waiting on the judge: "+record, "The knowledge base asked the judge for a part of this record and no answer has come. It asks again by itself; the tool knowledge_asked lists what is in flight.",
|
||||
record+" is judged again"); err != nil {
|
||||
_ = k.Store.Untold(ctx, partWaitPrefix+key)
|
||||
}
|
||||
}
|
||||
|
||||
// clearPartWaits clears the ask of every part that is no longer waiting.
|
||||
func (k *KB) clearPartWaits(live map[string]bool) {
|
||||
if k.Clear == nil || k.Store == nil {
|
||||
return
|
||||
}
|
||||
ctx := context.Background()
|
||||
told, err := k.Store.ToldOf(ctx, partWaitPrefix)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, t := range told {
|
||||
if live[strings.TrimPrefix(t, partWaitPrefix)] {
|
||||
continue
|
||||
}
|
||||
if err := k.Clear("knowledge."+t, "its part was judged"); err == nil {
|
||||
_ = k.Store.Untold(ctx, t)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// formRefused counts the records a form-only version would be refused for.
|
||||
func (k *KB) formRefused() int {
|
||||
done, err := k.FormOnly("", nil, false)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
n := 0
|
||||
for _, d := range done {
|
||||
if strings.HasPrefix(d.Skipped, "the check refused") {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||
)
|
||||
|
||||
// The health on the ledger (to-be 55 §1.1, AC56): it names where the records are served from and the move's last check, is
|
||||
// unhealthy for a ledger or a stream that does not answer, a head behind its history, an index behind the ledger and a stream
|
||||
// at its cap, and counts what waits on a person (an idle draft, a part waiting on the judge, a form-only check refused) without
|
||||
// failing; a part waiting more than 30 minutes is asked of the operator once, and the ask clears when it is judged.
|
||||
func TestTheHealthOnTheLedgerNamesTheSourceAndFailsForWhatTheLedgerOwes(t *testing.T) {
|
||||
w, f := ledgerWorld(t)
|
||||
ls := w.k.Store.(*ledgerStore)
|
||||
if err := ls.BuildIndex(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.fill = buscalls.StreamFilled{Fill: buscalls.Fill{Count: 3, Bytes: 100, MaxBytes: 512 << 20}}
|
||||
base := HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves"}
|
||||
ans := w.k.ledgerHealth(base)
|
||||
if !ans.Healthy || !strings.Contains(ans.Why, "served from the ledger") || !strings.Contains(ans.Why, "search index holds") {
|
||||
t.Fatalf("%+v", ans)
|
||||
}
|
||||
// Each thing the ledger owes, one at a time.
|
||||
for name, c := range map[string]struct {
|
||||
set func()
|
||||
want string
|
||||
}{
|
||||
"a head behind": {func() {
|
||||
f.lag = buscalls.Lagged{Instances: []buscalls.InstanceLag{{Instance: "decision/0290", History: 9, State: 7}}}
|
||||
}, "behind their history"},
|
||||
"a stream at a cap": {func() { f.fill.Bytes = f.fill.MaxBytes }, "at its cap"},
|
||||
"a stream that does not answer": {func() { f.noFill = true }, "does not answer"},
|
||||
} {
|
||||
f.lag, f.noFill = buscalls.Lagged{}, false
|
||||
f.fill = buscalls.StreamFilled{Fill: buscalls.Fill{Bytes: 100, MaxBytes: 512 << 20}}
|
||||
c.set()
|
||||
got := w.k.ledgerHealth(base)
|
||||
if got.Healthy || !strings.Contains(got.Why, c.want) {
|
||||
t.Errorf("%s: %+v", name, got)
|
||||
}
|
||||
}
|
||||
f.lag, f.noFill, f.fill = buscalls.Lagged{}, false, buscalls.StreamFilled{}
|
||||
// An index behind the ledger: a record the ledger holds and the index does not.
|
||||
ls.index.mu.Lock()
|
||||
ls.index.docsOf = map[Key][]string{}
|
||||
ls.index.mu.Unlock()
|
||||
if got := w.k.ledgerHealth(base); got.Healthy || !strings.Contains(got.Why, "behind the ledger") {
|
||||
t.Errorf("an index behind: %+v", got)
|
||||
}
|
||||
_ = ls.BuildIndex(t.Context())
|
||||
// What waits on a person is counted and leaves it healthy; a part waiting more than 30 minutes is asked once.
|
||||
w.k.States.State(askedState).Put("decision.0300.abc", map[string]any{"part": "Context", "record": "ADR 0300", "at": w.clock.Add(-time.Hour)})
|
||||
w.k.States.State(draftsState).Put("decision.0301", Draft{Kind: KindDecision, Number: 301, Author: "a@n", Touched: w.clock.AddDate(0, 0, -8)})
|
||||
got := w.k.ledgerHealth(base)
|
||||
if !got.Healthy || !strings.Contains(got.Why, "1 part(s) waiting") || !strings.Contains(got.Why, "1 draft(s) untouched") {
|
||||
t.Fatalf("%+v", got)
|
||||
}
|
||||
w.k.ledgerHealth(base)
|
||||
if len(w.told) != 1 {
|
||||
t.Fatalf("the operator was asked %d times", len(w.told))
|
||||
}
|
||||
w.k.States.State(askedState).Delete("decision.0300.abc")
|
||||
w.k.ledgerHealth(base)
|
||||
if len(w.cleared) != 1 {
|
||||
t.Fatalf("the ask was not cleared: %v", w.cleared)
|
||||
}
|
||||
}
|
||||
|
||||
// The records served from the database say so, and fail nothing.
|
||||
func TestTheHealthOnTheDatabaseSaysSo(t *testing.T) {
|
||||
db := newMemStore()
|
||||
states := newFakeStates()
|
||||
m := newMovableStore(db, newLedgerStore(newFakeLedger(t), states), states)
|
||||
k := &KB{Store: m, States: states}
|
||||
ans := k.ledgerHealth(HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves"})
|
||||
if !ans.Healthy || !strings.Contains(ans.Why, "served from the database") {
|
||||
t.Fatalf("%+v", ans)
|
||||
}
|
||||
}
|
||||
@@ -30,6 +30,10 @@ type fakeLedger struct {
|
||||
// failApply, when set, refuses the apply of that operation.
|
||||
failApply string
|
||||
account string
|
||||
// lag and fill are what lag and a stream's fill answer; noFill makes the stream not answer.
|
||||
lag buscalls.Lagged
|
||||
fill buscalls.StreamFilled
|
||||
noFill bool
|
||||
}
|
||||
|
||||
func newFakeLedger(t *testing.T) *fakeLedger {
|
||||
@@ -97,6 +101,13 @@ func (f *fakeLedger) Call(method string, req, answer any) error {
|
||||
page.Operations = append(page.Operations, o)
|
||||
}
|
||||
out = page
|
||||
case buscalls.LedgerLag:
|
||||
out = f.lag
|
||||
case buscalls.StreamFill:
|
||||
if f.noFill {
|
||||
return fmt.Errorf("the stream does not answer")
|
||||
}
|
||||
out = f.fill
|
||||
default:
|
||||
return fmt.Errorf("the fake ledger answers no %s", method)
|
||||
}
|
||||
|
||||
@@ -106,7 +106,7 @@ func serve(k *KB, status string) {
|
||||
k.kick = make(chan struct{}, 1)
|
||||
go hear(k)
|
||||
go background(k)
|
||||
if err := stdio.Serve("", append(append(Tools(k, callerOf), healthTool(status, k.now), highestTool(k), marksTool(k)), moduleTools(k, callerOf)...)); err != nil {
|
||||
if err := stdio.Serve("", append(append(Tools(k, callerOf), healthTool(status, k.now, k), highestTool(k), marksTool(k)), moduleTools(k, callerOf)...)); err != nil {
|
||||
logf("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
@@ -617,14 +617,20 @@ func (k *KB) Health(ready Healther) HealthAnswer {
|
||||
return HealthAnswer{State: "healthy", Healthy: true, Why: why}
|
||||
}
|
||||
|
||||
func healthTool(status string, now func() time.Time) stdio.Tool {
|
||||
func healthTool(status string, now func() time.Time, k *KB) stdio.Tool {
|
||||
return stdio.Tool{Name: HealthTool,
|
||||
Description: "How the knowledge base is, as its watch last saw it: healthy when its database serves this version, unhealthy " +
|
||||
"otherwise. A rule that judges the whole record set and fails — a decision accepted that nothing cites, a supersession " +
|
||||
"or a change said on one end only, a record extending a superseded one — is named but leaves it healthy: it is told " +
|
||||
"to the operator as a warning for the records' authors (novox/hq issue 504). So is a proposed record that ran out of " +
|
||||
"review tries: it waits on its author (novox/hq issue 508).",
|
||||
Run: func(map[string]any) (any, error) { return ReadStatus(status, now()), nil }}
|
||||
Run: func(map[string]any) (any, error) {
|
||||
ans := ReadStatus(status, now())
|
||||
if k != nil {
|
||||
ans = k.ledgerHealth(ans) // once the records are on the ledger, what the ledger form owes (health.go)
|
||||
}
|
||||
return ans, nil
|
||||
}}
|
||||
}
|
||||
|
||||
// ---- the verbs --------------------------------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user