diff --git a/modules/mesh-knowledge/README.md b/modules/mesh-knowledge/README.md index d5f82b1f..9172b2b4 100644 --- a/modules/mesh-knowledge/README.md +++ b/modules/mesh-knowledge/README.md @@ -92,7 +92,11 @@ record it goes out from. - A judgement that cannot be had (the judge unreachable, or an answer that is not `approve`, or `changes` naming no finding) leaves the record proposed and says so on it (`get`'s `review`). It is never taken as an approval. It is asked again after 10, then 20 minutes; after 3 tries the record says its author proposes it - again to have it reviewed again. + again to have it reviewed again, and the module is unhealthy, naming it, so the mesh raises a condition. + An ask of the operator that cannot be made counts as a try the same way. + - A review whose outcome the database refused to keep is held back for that record alone (10, then 20 + minutes, 3 tries, logged as given up), so it never blocks the records behind it. + - A progressive insight waiting for its review is revised only by its proven author. - A record revised while it is judged is not decided on the version judged; the newest is judged next. - On approve the record and its clauses are accepted and in force. Every clause it changes moves to `changed`, and every record it supersedes moves to `superseded`, each as a new version of that record. A record marked `needs-operator` is first asked of the diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go b/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go index f4a48506..64b12530 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go @@ -83,6 +83,9 @@ type KB struct { // mu serialises the writes: one instance holds the seat, and a write reads, judges and writes as one. mu sync.Mutex + stuckMu sync.Mutex + stuck map[Key]*stuckReview + // kick wakes the background work when a proposal waits for its review; nil where nothing runs in the // background (the tests take each step themselves). kick chan struct{} @@ -321,6 +324,10 @@ func (k *KB) Propose(caller string, a ProposeAsk) (*Proposed, error) { why = "revised" } case held.Current.Accepted(): + if held.Pending != nil && (!proven(caller) || caller != held.Pending.Author) { + return nil, fmt.Errorf("%s has a progressive insight by %s waiting for its review: only its author, proven by the "+ + "runtime, revises it (this call is by %s). Nothing was proposed", key, held.Pending.Author, caller) + } if strings.TrimSpace(a.Insight) == "" { return nil, fmt.Errorf("%s is %s: it changes only by a new record, or by a progressive insight, a correction of fact "+ "that leaves the decision standing (give insight: what stood before). Nothing was proposed", key, held.Current.Status) diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go index 347175ac..f0342f98 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go @@ -474,8 +474,9 @@ type HealthAnswer struct { At time.Time `json:"at,omitempty"` } -// Health is how the knowledge base is: unhealthy when its database cannot serve, or when a rule that judges the -// whole record set fails — the mesh raises a condition naming the rule and the records. +// Health is how the knowledge base is: unhealthy when its database cannot serve, when a rule that judges the +// whole record set fails, or when a proposed record ran out of review tries — the mesh raises a condition naming +// the rule, the records, or the records not reviewed. func (k *KB) Health(ready Healther) HealthAnswer { ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) defer cancel() @@ -488,25 +489,42 @@ func (k *KB) Health(ready Healther) HealthAnswer { if err != nil { return HealthAnswer{State: "unhealthy", Why: "the whole-set findings cannot be read: " + err.Error()} } - if len(findings) > 0 { - var says []string - for _, f := range findings { - says = append(says, "the rule "+f.Rule+": "+f.Says) - } - sort.Strings(says) - if len(says) > 8 { - says = append(says[:8], fmt.Sprintf("and %d more", len(says)-8)) - } - return HealthAnswer{State: "unhealthy", Why: fmt.Sprintf("%d rule(s) that judge the whole record set fail: %s", len(findings), strings.Join(says, "; "))} + var says []string + for _, f := range findings { + says = append(says, "the rule "+f.Rule+": "+f.Says) } - return HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves, and every rule that judges the whole record set holds"} + sort.Strings(says) + all, err := k.Store.All(ctx) + if err != nil { + return HealthAnswer{State: "unhealthy", Why: "the records cannot be read: " + err.Error()} + } + unreviewed := Unreviewed(all) + if len(findings)+len(unreviewed) > 0 { + var parts []string + if len(unreviewed) > 0 { + if len(unreviewed) > 8 { + unreviewed = append(unreviewed[:8], fmt.Sprintf("and %d more", len(unreviewed)-8)) + } + parts = append(parts, fmt.Sprintf("%d proposed record(s) could not be reviewed in %d tries and wait for their authors to propose them again: %s", + len(Unreviewed(all)), MostReviewTries, strings.Join(unreviewed, "; "))) + } + if len(says) > 0 { + if len(says) > 8 { + says = append(says[:8], fmt.Sprintf("and %d more", len(says)-8)) + } + parts = append(parts, fmt.Sprintf("%d rule(s) that judge the whole record set fail: %s", len(findings), strings.Join(says, "; "))) + } + return HealthAnswer{State: "unhealthy", Why: strings.Join(parts, ". ")} + } + return HealthAnswer{State: "healthy", Healthy: true, Why: "the database serves, every rule that judges the whole record set holds, and no proposal ran out of review tries"} } func healthTool(status string, now func() time.Time) 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 and every " + "rule that judges the whole record set holds; unhealthy otherwise, naming the rule and the records — a decision accepted " + - "that nothing cites, a supersession or a change said on one end only, a record extending a superseded one.", + "that nothing cites, a supersession or a change said on one end only, a record extending a superseded one — or when a " + + "proposed record ran out of review tries, naming it.", Run: func(map[string]any) (any, error) { return ReadStatus(status, now()), nil }} } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go b/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go index b99d9596..11c5798c 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go @@ -131,17 +131,23 @@ func newAskID(r *Record) string { } // askOperator asks the operator to approve a record the knowledge base's review approved, and keeps the ask with -// it. The caller holds the write lock. An ask that could not be published is taken back, and the review is asked -// again after a pause: the record stays proposed. -func (k *KB) askOperator(ctx context.Context, held *Entry, w *Record, by string) (string, error) { +// it. The caller holds the write lock. An ask that could not be made or published is taken back, and the review is +// asked again after a pause, counted with the tries before it (tries): the record stays proposed, and is reviewed +// at most MostReviewTries times. +func (k *KB) askOperator(ctx context.Context, held *Entry, w *Record, by string, tries int) (string, error) { version := held.CurrentVersion if held.Pending != nil { version = held.PendingVersion } unasked := func(why string) (string, error) { w.Ask = nil - w.Judging = &Judging{State: JudgingFailed, Tries: 1, Reason: why, At: k.now()} - said := "approved by the knowledge base's review, and " + why + "; it stays proposed and is reviewed again" + w.Judging = &Judging{State: JudgingFailed, Tries: tries + 1, Reason: why, At: k.now()} + said := fmt.Sprintf("approved by the knowledge base's review, and %s (try %d of %d); it stays proposed", why, tries+1, MostReviewTries) + if tries+1 >= MostReviewTries { + said += ", and its author proposes it again to have it reviewed again" + } else { + said += " and is reviewed again after a pause" + } e, err := k.Store.Get(ctx, w.Key()) if err != nil || e == nil { return "", fmt.Errorf("%s: %s, and that could not be kept (%v)", w.Key(), said, err) diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/review.go b/modules/mesh-knowledge/cmd/mesh-knowledge/review.go index 603d8115..c864f397 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/review.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/review.go @@ -156,8 +156,48 @@ func (k *KB) reviewDue(e *Entry) bool { return false } -// ReviewStep reviews the first proposed version waiting for its review. It answers what it did, or "" when -// nothing waits. +// stuckReview is a record whose review could not be kept — the store refused the write, or the acceptance +// failed — so its state never reached the store: held here, per record and per revision, so it is tried again +// only after a pause, at most MostReviewTries times, and never blocks the records behind it. +type stuckReview struct { + revision int64 + tries int + at time.Time +} + +// held says whether a record's review waits here, after a failure the store could not keep. +func (k *KB) heldBack(e *Entry) bool { + k.stuckMu.Lock() + defer k.stuckMu.Unlock() + s := k.stuck[e.Current.Key()] + if s == nil || s.revision != e.Revision { + return false + } + return s.tries >= MostReviewTries || k.now().Sub(s.at) < time.Duration(s.tries)*10*time.Minute +} + +func (k *KB) noteStuck(e *Entry, err error) { + k.stuckMu.Lock() + defer k.stuckMu.Unlock() + if k.stuck == nil { + k.stuck = map[Key]*stuckReview{} + } + key := e.Current.Key() + s := k.stuck[key] + if s == nil || s.revision != e.Revision { + s = &stuckReview{revision: e.Revision} + k.stuck[key] = s + } + s.tries++ + s.at = k.now() + if s.tries >= MostReviewTries { + k.logf("[mesh-knowledge] REVIEW GIVEN UP: the review of %s could not be kept %d times (last: %v); it stays proposed", key, s.tries, err) + } +} + +// ReviewStep reviews the first proposed version waiting for its review and not held back. It answers what it +// did, or "" when nothing waits; an error is one record's, which is then held back, so the next step takes the +// next record. func (k *KB) ReviewStep() (string, error) { if k.Judge == nil { return "", nil @@ -167,13 +207,35 @@ func (k *KB) ReviewStep() (string, error) { return "", err } for _, e := range all { - if k.reviewDue(e) { - return k.reviewOne(e.Current.Key()) + if !k.reviewDue(e) || k.heldBack(e) { + continue } + said, err := k.reviewOne(e.Current.Key()) + if err != nil { + k.noteStuck(e, err) + return "", err + } + k.stuckMu.Lock() + delete(k.stuck, e.Current.Key()) + k.stuckMu.Unlock() + return said, nil } return "", nil } +// Unreviewed are the proposed records that ran out of review tries: no judgement could be had, or the operator +// could not be asked. They make the module unhealthy, so the mesh raises a condition naming them. +func Unreviewed(all []*Entry) []string { + var out []string + for _, e := range all { + w := e.Working() + if w.Status == StatusProposed && w.Judging != nil && w.Judging.State == JudgingFailed && w.Judging.Tries >= MostReviewTries { + out = append(out, fmt.Sprintf("%s (%s)", w.Key(), w.Judging.Reason)) + } + } + return out +} + // reviewOne asks the judgement on one record's working version, and acts on its verdict. The write lock is held // only to read and to write, never across the judgement; a record revised while it was judged is judged again. func (k *KB) reviewOne(key Key) (string, error) { @@ -252,6 +314,7 @@ func (k *KB) reviewOne(key Key) (string, error) { k.emit(EventReviewed, w, by, why, now.Versions+1) return key.String() + ": " + why, nil } + tries := w.Judging.Tries w.Judging = nil w.Reviews = append(w.Reviews, Review{By: by, At: k.now(), Verdict: verdict, Findings: strings.Join(findings, "\n"), Version: version}) if verdict == VerdictChanges { @@ -263,7 +326,7 @@ func (k *KB) reviewOne(key Key) (string, error) { return key.String() + ": " + why, nil } if w.NeedsOperator { - return k.askOperator(ctx, now, w, by) + return k.askOperator(ctx, now, w, by, tries) } k.emit(EventReviewed, w, by, "approved", now.Versions+1) r, err := k.accept(ctx, now, w, by, "accepted after the knowledge base's review", "") diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go new file mode 100644 index 00000000..974d5635 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/review_bounds_test.go @@ -0,0 +1,119 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + "time" +) + +// The review of mesh-catalog#211: every way a review can fail is bounded, per record, and loud. + +// A needs-operator record whose ask cannot be made is reviewed again at most MostReviewTries times, however long +// the clock runs: the tries are carried, not started again at each failure. +func TestAnAskThatCannotBeMadeIsTriedABoundedNumberOfTimes(t *testing.T) { + w := imported(t, newMemStore()) + w.k.Asks = nil + n := proposed(t, w, "author@g14", true) + for i := 0; i < 6*48; i++ { // two days, every ten minutes + w.reviewAll(t) + w.clock = w.clock.Add(10 * time.Minute) + } + e := mustGet(t, w.k.Store, Key{KindDecision, n}) + if len(w.judge.reviews) != MostReviewTries || e.Current.Judging == nil || e.Current.Judging.Tries != MostReviewTries || + len(e.Current.Reviews) > MostReviewTries || e.Current.Status != StatusProposed { + t.Fatalf("%d judgements, judging %+v, %d reviews kept", len(w.judge.reviews), e.Current.Judging, len(e.Current.Reviews)) + } +} + +// failingSaves is a store whose writes of one record fail, as a database refusing them would. +type failingSaves struct { + *memStore + key Key + fail bool +} + +func (f *failingSaves) Save(ctx context.Context, v Version, current, clearPending bool, revision int64) error { + if f.fail && v.Record != nil && v.Record.Key() == f.key { + return errors.New("the database refused the write") + } + return f.memStore.Save(ctx, v, current, clearPending, revision) +} + +// A record whose review cannot be kept does not block the others, and is tried a bounded number of times. +func TestARecordWhoseReviewCannotBeKeptBlocksNoOtherAndIsTriedBoundedly(t *testing.T) { + s := &failingSaves{memStore: newMemStore()} + w := imported(t, s) + a := proposed(t, w, "author@g14", false) + b := proposed(t, w, "author@g14", false) + s.key, s.fail = Key{KindDecision, a}, true + errs := 0 + for i := 0; i < 6*48; i++ { + for j := 0; j < 5; j++ { + said, err := w.k.ReviewStep() + if err != nil { + errs++ + continue + } + if said == "" { + break + } + } + w.clock = w.clock.Add(10 * time.Minute) + } + if got := mustGet(t, s, Key{KindDecision, b}).Current.Status; got != StatusAccepted { + t.Fatalf("ADR %04d stayed %s behind a record whose review could not be kept", b, got) + } + if errs == 0 || errs > MostReviewTries { + t.Fatalf("the failing record was tried %d times; at most %d", errs, MostReviewTries) + } + if mustGet(t, s, Key{KindDecision, a}).Current.Status != StatusProposed { + t.Fatal("accepted without its review kept") + } +} + +// A record that ran out of review tries makes the module unhealthy, naming it: the mesh raises a condition. +func TestARecordThatCouldNotBeReviewedMakesTheModuleUnhealthy(t *testing.T) { + w := imported(t, newMemStore()) + n := proposed(t, w, "author@g14", false) + w.judge.reviewFail = true + for i := 0; i < MostReviewTries; i++ { + w.reviewAll(t) + if h := w.k.Health(nil); i < MostReviewTries-1 && !h.Healthy { + t.Fatalf("unhealthy while tries remain: %s", h.Why) + } + w.clock = w.clock.Add(time.Hour) + } + h := w.k.Health(nil) + if h.Healthy || !strings.Contains(h.Why, itoaKey(n)) || !strings.Contains(h.Why, "could not be reviewed") { + t.Fatalf("%+v", h) + } +} + +// A progressive insight waiting for its review is replaced only by its author, proven. +func TestAPendingInsightIsReplacedOnlyByItsAuthor(t *testing.T) { + w := imported(t, newMemStore()) + n := w.reserved(t, "author@g14") + w.accepted(t, "author@g14", valid(n)) + p := valid(n) + p.Sections[2].Text = "Merges land sooner, by about a minute." + p.Insight = "it said only that merges land sooner" + if _, err := w.k.Propose("corrector@ace", p); err != nil { + t.Fatal(err) + } + q := valid(n) + q.Sections[2].Text = "Merges land later." + q.Insight = "it said merges land sooner" + for _, other := range []string{"author@g14", "other@novox", Unproven} { + if _, err := w.k.Propose(other, q); err == nil || !strings.Contains(err.Error(), "only its author") { + t.Fatalf("a pending insight replaced by %s: %v", other, err) + } + } + if e := mustGet(t, w.k.Store, Key{KindDecision, n}); e.Pending.Section("Consequences") != p.Sections[2].Text { + t.Fatalf("the pending insight was replaced: %q", e.Pending.Section("Consequences")) + } + if _, err := w.k.Propose("corrector@ace", q); err != nil { + t.Fatalf("its author's revision: %v", err) + } +}