mesh-knowledge: bound every review failure per record, make one out of tries loud, and let only its author revise a pending insight (review of #211)

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)

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 }}
 }
 

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)

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", "")

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)
+	}
+}