Merge pull request 'Brake a withdrawal larger than its bound (hq to-be 45 Phase 2, ADR 0227 rule 4)' (#90) from feat/a-core-that-cannot-fail-silently-phase-2 into main

modules/keycloak/cmd/keycloak-provider/harness.go

@@ -1,7 +1,7 @@
 package main
 
 // The reconcile loop every provider shares, as the TypeScript SDK's runProvisioner runs it
-// (@novox/mesh-sdk/provisioner, 0.1.10). The Go SDK has no provisioner yet, so this module carries
+// (@novox/mesh-sdk/provisioner, 0.1.11). The Go SDK has no provisioner yet, so this module carries
 // the loop itself, line for line in behaviour; when the Go SDK grows one, this file is what moves
 // there (novox/hq ADR 0039: the loop is the SDK's, the adapter is the module's).
 //
@@ -17,6 +17,16 @@ package main
 // identity provider failed every consumer 31,000 times in a day and said so only in its journal
 // (novox/hq issue 179).
 //
+// **A reconcile that would withdraw more than its bound stops, and says so (novox/hq to-be 45 Phase 2,
+// ADR 0227 rule 4).** Withdrawing more than WithdrawAtOnce consumers in one pass — or more than
+// WithdrawFraction of those this process holds — is the shape of issue 241, where one misread file
+// withdrew seven at once. Such a pass withdraws nothing: each consumer it would have withdrawn is kept,
+// announced `provisioner.failing` with the class `withdrawal-braked` (the controller raises it as a
+// condition), and said. While the same consumers stay unasked for, one is released every ReleaseEvery,
+// said and announced as it goes, so an intended unassignment of many completes without a hand and a
+// mistaken one costs at most one consumer an hour while the operator is told. Withdrawal never destroys
+// data (issue 241's second half), so a release is a login locked, not a database dropped.
+//
 // Carried, identical, by every Go provider until the Go SDK has the loop: postgres and keycloak.
 // Each module's `harness_same_test.go` fails when its copy and the other's differ.
 
@@ -27,6 +37,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
 	"time"
 )
@@ -74,8 +85,19 @@ type Harness struct {
 	// missed the first hears the next, and a standing nobody repeats can be told from one that holds.
 	FailingAfter  time.Duration
 	SayAgainEvery time.Duration
+	// WithdrawAtOnce (1) and WithdrawFraction (0.5) bound what one pass may withdraw: more consumers
+	// than WithdrawAtOnce, or a larger share of those held than WithdrawFraction, brakes the pass.
+	// ReleaseEvery (1h) is how often a braked withdrawal lets one consumer go.
+	WithdrawAtOnce   int
+	WithdrawFraction float64
+	ReleaseEvery     time.Duration
 
-	verifiedAt  time.Time
+	verifiedAt time.Time
+	// braked is every consumer a braked pass kept, by when it was first kept; releasedAt is when the
+	// brake last let one go, and brakeSaid the set it last said, so a pass repeats nothing.
+	braked      map[string]time.Time
+	releasedAt  time.Time
+	brakeSaid   string
 	applied     map[string]appliedEntry
 	lost        map[string]brake
 	waiting     map[string]int
@@ -109,6 +131,9 @@ const (
 	ClassUnreachable = "unreachable"
 	ClassSecret      = "secret-unreadable"
 	ClassRefused     = "refused"
+	// ClassWithdrawalBraked is a consumer the mesh no longer asks for, kept because the pass that would
+	// withdraw it would withdraw more than its bound (ADR 0227 rule 4).
+	ClassWithdrawalBraked = "withdrawal-braked"
 )
 
 // Classifier is an adapter that can say what class an error of its own is.
@@ -119,6 +144,7 @@ type Classifier interface {
 type appliedEntry struct {
 	hash    string
 	derived map[string]any
+	node    string
 }
 
 type brake struct {
@@ -170,6 +196,15 @@ func (h *Harness) init() {
 	if h.SayAgainEvery == 0 {
 		h.SayAgainEvery = 15 * time.Minute
 	}
+	if h.WithdrawAtOnce == 0 {
+		h.WithdrawAtOnce = 1
+	}
+	if h.WithdrawFraction == 0 {
+		h.WithdrawFraction = 0.5
+	}
+	if h.ReleaseEvery == 0 {
+		h.ReleaseEvery = time.Hour
+	}
 	if h.Log == nil {
 		h.Log = func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) }
 	}
@@ -180,6 +215,7 @@ func (h *Harness) init() {
 		h.failing = map[string]failure{}
 		h.trouble = map[string]*standing{}
 		h.cleared = map[string]bool{}
+		h.braked = map[string]time.Time{}
 	}
 }
 
@@ -359,7 +395,7 @@ func (h *Harness) Reconcile(ctx context.Context) {
 			delete(h.failing, g.As)
 		}
 		h.succeeded(g.As)
-		h.applied[g.As] = appliedEntry{hash: hash, derived: p.Derived}
+		h.applied[g.As] = appliedEntry{hash: hash, derived: p.Derived, node: g.Node}
 		if reapplying == 0 {
 			delete(h.lost, g.As)
 		} else {
@@ -374,11 +410,29 @@ func (h *Harness) Reconcile(ctx context.Context) {
 		}
 	}
 
-	// Withdraw every login this process made that the mesh no longer asks for.
-	for as, was := range h.applied {
-		if want[as] {
-			continue
+	// Withdraw every login this process made that the mesh no longer asks for — within the bound.
+	var withdrawing []string
+	for as := range h.applied {
+		if !want[as] {
+			withdrawing = append(withdrawing, as)
 		}
+	}
+	sort.Strings(withdrawing)
+	for as := range h.braked {
+		if want[as] {
+			// Asked for again: the brake held what the mesh still wanted. Its standing was ended by this
+			// pass's success above, as any consumer's is.
+			delete(h.braked, as)
+		}
+	}
+	if h.overTheBound(len(withdrawing), len(h.applied)) {
+		withdrawing = h.brakeWithdrawal(withdrawing)
+	} else if len(h.braked) > 0 {
+		h.say("the withdrawal is within its bound again: %s withdrawn as asked", strings.Join(withdrawing, ", "))
+		h.braked, h.brakeSaid = map[string]time.Time{}, ""
+	}
+	for _, as := range withdrawing {
+		was := h.applied[as]
 		h.say("%s: no longer in %s; withdrawing it from the backend", as, h.Receives)
 		if err := h.Adapter.Remove(ctx, as, was.derived); err != nil {
 			h.say("%s: remove failed, will retry: %v", as, err)
@@ -386,6 +440,7 @@ func (h *Harness) Reconcile(ctx context.Context) {
 		}
 		delete(h.applied, as)
 		delete(h.lost, as)
+		delete(h.braked, as)
 	}
 	for as := range h.failing {
 		if !want[as] {
@@ -393,14 +448,59 @@ func (h *Harness) Reconcile(ctx context.Context) {
 		}
 	}
 	// A consumer the mesh stopped asking for is no longer failed by anyone: said, so a standing
-	// the controller keeps for it is cleared rather than left naming a consumer that is gone.
+	// the controller keeps for it is cleared rather than left naming a consumer that is gone. One the
+	// brake holds is still kept, and its standing stays.
 	for as := range h.trouble {
-		if !want[as] {
+		if _, held := h.braked[as]; !want[as] && !held {
 			h.recovered(as, "withdrawn")
 		}
 	}
 }
 
+// overTheBound says a pass withdrawing n of the held consumers would withdraw more than it may.
+func (h *Harness) overTheBound(n, held int) bool {
+	if n == 0 {
+		return false
+	}
+	return n > h.WithdrawAtOnce || (held > 1 && float64(n) > h.WithdrawFraction*float64(held))
+}
+
+// brakeWithdrawal keeps every consumer a pass over its bound would withdraw, announces each as failing
+// with the class withdrawal-braked, and answers the one it releases now, if one is due.
+func (h *Harness) brakeWithdrawal(withdrawing []string) []string {
+	now := h.Now()
+	set := strings.Join(withdrawing, ", ")
+	if set != h.brakeSaid {
+		h.say("WITHDRAWAL BRAKED: this pass would withdraw %d of the %d consumer(s) this provider holds (%s), "+
+			"more than %d at once or %.0f%% of them. Nothing is withdrawn; each is announced as %s (%s), and one "+
+			"is let go every %s while the mesh goes on not asking for them (novox/hq ADR 0227 rule 4)",
+			len(withdrawing), len(h.applied), set, h.WithdrawAtOnce, h.WithdrawFraction*100, EventFailing,
+			ClassWithdrawalBraked, h.ReleaseEvery)
+		h.brakeSaid = set
+	}
+	if len(h.braked) == 0 {
+		// The release clock starts with the brake, not at the last release of an earlier one.
+		h.releasedAt = now
+	}
+	for _, as := range withdrawing {
+		if _, kept := h.braked[as]; !kept {
+			h.braked[as] = now
+		}
+		text := fmt.Sprintf("the mesh no longer asks for it, and the pass that would withdraw it would withdraw %d "+
+			"consumers at once: kept until released (%s)", len(withdrawing), set)
+		h.failed(as, h.applied[as].node, ClassWithdrawalBraked, text)
+	}
+	if now.Sub(h.releasedAt) < h.ReleaseEvery {
+		return nil
+	}
+	h.releasedAt = now
+	release := withdrawing[0]
+	h.say("%s: released by the withdrawal brake after %s; %d more kept", release,
+		now.Sub(h.braked[release]).Round(time.Second), len(withdrawing)-1)
+	h.recovered(release, "withdrawn")
+	return []string{release}
+}
+
 // failed counts one more failure in a consumer's unbroken run, and announces the run once it has
 // lasted FailingAfter — then again every SayAgainEvery while it lasts.
 func (h *Harness) failed(as, node, class, text string) {

modules/postgres/cmd/postgres-provider/brake_test.go

@@ -0,0 +1,113 @@
+package main
+
+// The withdrawal brake (novox/hq to-be 45 Phase 2, ADR 0227 rule 4; replay R6 of issue 241): a pass that
+// would withdraw more than its bound withdraws nothing, says so, and announces every consumer it kept as
+// failing with the class withdrawal-braked, which the controller raises as a condition; one is let go
+// every hour while the mesh goes on not asking; a consumer asked for again is kept and said recovered.
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+// sevenConsumers is the control node's postgres on 2026-10-04: seven databases, all in use.
+func sevenConsumers(w *world) {
+	var given []map[string]any
+	for i := 1; i <= 7; i++ {
+		given = append(given, map[string]any{"as": fmt.Sprintf("consumer%d", i), "node": "anchor"})
+	}
+	w.give(given...)
+	w.h.Reconcile(ctx)
+}
+
+func TestAWithdrawalOfEveryConsumerIsBrakedAndSaid(t *testing.T) {
+	w, said := standingWorld(t)
+	sevenConsumers(w)
+	// A file read whole that names nobody: the shape of 241 that its first fix does not catch.
+	w.give()
+	w.passes(6 * time.Minute)
+	if len(w.a.removed) != 0 {
+		t.Fatalf("a pass over its bound withdrew %v", w.a.removed)
+	}
+	braked := 0
+	for _, a := range *said {
+		if a.event == EventFailing && a.body["class"] == ClassWithdrawalBraked && a.body["node"] == "anchor" {
+			braked++
+		}
+	}
+	if braked != 7 {
+		t.Fatalf("%d of the 7 consumers kept were announced as braked: %v", braked, *said)
+	}
+	if !strings.Contains(strings.Join(w.said, "\n"), "WITHDRAWAL BRAKED") {
+		t.Fatal("the brake was not said")
+	}
+
+	// One is let go once the brake has held an hour, and then one an hour.
+	w.passes(55 * time.Minute)
+	if len(w.a.removed) != 1 {
+		t.Fatalf("after an hour of the mesh not asking, %d were withdrawn, want 1: %v", len(w.a.removed), w.a.removed)
+	}
+	w.passes(59 * time.Minute)
+	if len(w.a.removed) != 1 {
+		t.Fatalf("a second was let go within the hour: %v", w.a.removed)
+	}
+	w.passes(2 * time.Minute)
+	if len(w.a.removed) != 2 {
+		t.Fatalf("the second was not let go after another hour: %v", w.a.removed)
+	}
+}
+
+func TestAConsumerAskedForAgainIsKeptAndNotWithdrawn(t *testing.T) {
+	w, said := standingWorld(t)
+	sevenConsumers(w)
+	w.give()
+	w.passes(6 * time.Minute)
+	// The file is right again: every consumer is asked for, nothing was withdrawn, each is recovered.
+	sevenConsumers(w)
+	w.passes(5 * time.Second)
+	if len(w.a.removed) != 0 {
+		t.Fatalf("withdrew %v", w.a.removed)
+	}
+	recovered := 0
+	for _, a := range *said {
+		if a.event == EventRecovered && a.body["why"] == nil {
+			recovered++
+		}
+	}
+	if recovered != 7 {
+		t.Fatalf("%d of the 7 kept consumers were said recovered: %v", recovered, *said)
+	}
+	if len(w.h.braked) != 0 {
+		t.Fatalf("the brake still holds %v", w.h.braked)
+	}
+}
+
+// Within the bound — one consumer of several, or the last one held — a withdrawal goes as asked.
+func TestAWithdrawalWithinItsBoundIsNotBraked(t *testing.T) {
+	w := newWorld(t)
+	w.give(map[string]any{"as": "a"}, map[string]any{"as": "b"}, map[string]any{"as": "c"})
+	w.h.Reconcile(ctx)
+	w.give(map[string]any{"as": "a"}, map[string]any{"as": "b"})
+	w.h.Reconcile(ctx)
+	if strings.Join(w.a.removed, ",") != "c" {
+		t.Fatalf("one of three was not withdrawn: %v", w.a.removed)
+	}
+	// Two of the remaining two at once is over the bound.
+	w.give()
+	w.h.Reconcile(ctx)
+	if strings.Join(w.a.removed, ",") != "c" {
+		t.Fatalf("two at once were withdrawn: %v", w.a.removed)
+	}
+	for _, c := range []struct{ n, held int }{{1, 1}, {1, 2}, {1, 3}} {
+		if w.h.overTheBound(c.n, c.held) {
+			t.Errorf("%d of %d is over the bound", c.n, c.held)
+		}
+	}
+	for _, c := range []struct{ n, held int }{{2, 2}, {2, 7}, {7, 7}} {
+		if !w.h.overTheBound(c.n, c.held) {
+			t.Errorf("%d of %d is within the bound", c.n, c.held)
+		}
+	}
+}

modules/postgres/cmd/postgres-provider/harness.go

@@ -1,7 +1,7 @@
 package main
 
 // The reconcile loop every provider shares, as the TypeScript SDK's runProvisioner runs it
-// (@novox/mesh-sdk/provisioner, 0.1.10). The Go SDK has no provisioner yet, so this module carries
+// (@novox/mesh-sdk/provisioner, 0.1.11). The Go SDK has no provisioner yet, so this module carries
 // the loop itself, line for line in behaviour; when the Go SDK grows one, this file is what moves
 // there (novox/hq ADR 0039: the loop is the SDK's, the adapter is the module's).
 //
@@ -17,6 +17,16 @@ package main
 // identity provider failed every consumer 31,000 times in a day and said so only in its journal
 // (novox/hq issue 179).
 //
+// **A reconcile that would withdraw more than its bound stops, and says so (novox/hq to-be 45 Phase 2,
+// ADR 0227 rule 4).** Withdrawing more than WithdrawAtOnce consumers in one pass — or more than
+// WithdrawFraction of those this process holds — is the shape of issue 241, where one misread file
+// withdrew seven at once. Such a pass withdraws nothing: each consumer it would have withdrawn is kept,
+// announced `provisioner.failing` with the class `withdrawal-braked` (the controller raises it as a
+// condition), and said. While the same consumers stay unasked for, one is released every ReleaseEvery,
+// said and announced as it goes, so an intended unassignment of many completes without a hand and a
+// mistaken one costs at most one consumer an hour while the operator is told. Withdrawal never destroys
+// data (issue 241's second half), so a release is a login locked, not a database dropped.
+//
 // Carried, identical, by every Go provider until the Go SDK has the loop: postgres and keycloak.
 // Each module's `harness_same_test.go` fails when its copy and the other's differ.
 
@@ -27,6 +37,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"sort"
 	"strings"
 	"time"
 )
@@ -74,8 +85,19 @@ type Harness struct {
 	// missed the first hears the next, and a standing nobody repeats can be told from one that holds.
 	FailingAfter  time.Duration
 	SayAgainEvery time.Duration
+	// WithdrawAtOnce (1) and WithdrawFraction (0.5) bound what one pass may withdraw: more consumers
+	// than WithdrawAtOnce, or a larger share of those held than WithdrawFraction, brakes the pass.
+	// ReleaseEvery (1h) is how often a braked withdrawal lets one consumer go.
+	WithdrawAtOnce   int
+	WithdrawFraction float64
+	ReleaseEvery     time.Duration
 
-	verifiedAt  time.Time
+	verifiedAt time.Time
+	// braked is every consumer a braked pass kept, by when it was first kept; releasedAt is when the
+	// brake last let one go, and brakeSaid the set it last said, so a pass repeats nothing.
+	braked      map[string]time.Time
+	releasedAt  time.Time
+	brakeSaid   string
 	applied     map[string]appliedEntry
 	lost        map[string]brake
 	waiting     map[string]int
@@ -109,6 +131,9 @@ const (
 	ClassUnreachable = "unreachable"
 	ClassSecret      = "secret-unreadable"
 	ClassRefused     = "refused"
+	// ClassWithdrawalBraked is a consumer the mesh no longer asks for, kept because the pass that would
+	// withdraw it would withdraw more than its bound (ADR 0227 rule 4).
+	ClassWithdrawalBraked = "withdrawal-braked"
 )
 
 // Classifier is an adapter that can say what class an error of its own is.
@@ -119,6 +144,7 @@ type Classifier interface {
 type appliedEntry struct {
 	hash    string
 	derived map[string]any
+	node    string
 }
 
 type brake struct {
@@ -170,6 +196,15 @@ func (h *Harness) init() {
 	if h.SayAgainEvery == 0 {
 		h.SayAgainEvery = 15 * time.Minute
 	}
+	if h.WithdrawAtOnce == 0 {
+		h.WithdrawAtOnce = 1
+	}
+	if h.WithdrawFraction == 0 {
+		h.WithdrawFraction = 0.5
+	}
+	if h.ReleaseEvery == 0 {
+		h.ReleaseEvery = time.Hour
+	}
 	if h.Log == nil {
 		h.Log = func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) }
 	}
@@ -180,6 +215,7 @@ func (h *Harness) init() {
 		h.failing = map[string]failure{}
 		h.trouble = map[string]*standing{}
 		h.cleared = map[string]bool{}
+		h.braked = map[string]time.Time{}
 	}
 }
 
@@ -359,7 +395,7 @@ func (h *Harness) Reconcile(ctx context.Context) {
 			delete(h.failing, g.As)
 		}
 		h.succeeded(g.As)
-		h.applied[g.As] = appliedEntry{hash: hash, derived: p.Derived}
+		h.applied[g.As] = appliedEntry{hash: hash, derived: p.Derived, node: g.Node}
 		if reapplying == 0 {
 			delete(h.lost, g.As)
 		} else {
@@ -374,11 +410,29 @@ func (h *Harness) Reconcile(ctx context.Context) {
 		}
 	}
 
-	// Withdraw every login this process made that the mesh no longer asks for.
-	for as, was := range h.applied {
-		if want[as] {
-			continue
+	// Withdraw every login this process made that the mesh no longer asks for — within the bound.
+	var withdrawing []string
+	for as := range h.applied {
+		if !want[as] {
+			withdrawing = append(withdrawing, as)
 		}
+	}
+	sort.Strings(withdrawing)
+	for as := range h.braked {
+		if want[as] {
+			// Asked for again: the brake held what the mesh still wanted. Its standing was ended by this
+			// pass's success above, as any consumer's is.
+			delete(h.braked, as)
+		}
+	}
+	if h.overTheBound(len(withdrawing), len(h.applied)) {
+		withdrawing = h.brakeWithdrawal(withdrawing)
+	} else if len(h.braked) > 0 {
+		h.say("the withdrawal is within its bound again: %s withdrawn as asked", strings.Join(withdrawing, ", "))
+		h.braked, h.brakeSaid = map[string]time.Time{}, ""
+	}
+	for _, as := range withdrawing {
+		was := h.applied[as]
 		h.say("%s: no longer in %s; withdrawing it from the backend", as, h.Receives)
 		if err := h.Adapter.Remove(ctx, as, was.derived); err != nil {
 			h.say("%s: remove failed, will retry: %v", as, err)
@@ -386,6 +440,7 @@ func (h *Harness) Reconcile(ctx context.Context) {
 		}
 		delete(h.applied, as)
 		delete(h.lost, as)
+		delete(h.braked, as)
 	}
 	for as := range h.failing {
 		if !want[as] {
@@ -393,14 +448,59 @@ func (h *Harness) Reconcile(ctx context.Context) {
 		}
 	}
 	// A consumer the mesh stopped asking for is no longer failed by anyone: said, so a standing
-	// the controller keeps for it is cleared rather than left naming a consumer that is gone.
+	// the controller keeps for it is cleared rather than left naming a consumer that is gone. One the
+	// brake holds is still kept, and its standing stays.
 	for as := range h.trouble {
-		if !want[as] {
+		if _, held := h.braked[as]; !want[as] && !held {
 			h.recovered(as, "withdrawn")
 		}
 	}
 }
 
+// overTheBound says a pass withdrawing n of the held consumers would withdraw more than it may.
+func (h *Harness) overTheBound(n, held int) bool {
+	if n == 0 {
+		return false
+	}
+	return n > h.WithdrawAtOnce || (held > 1 && float64(n) > h.WithdrawFraction*float64(held))
+}
+
+// brakeWithdrawal keeps every consumer a pass over its bound would withdraw, announces each as failing
+// with the class withdrawal-braked, and answers the one it releases now, if one is due.
+func (h *Harness) brakeWithdrawal(withdrawing []string) []string {
+	now := h.Now()
+	set := strings.Join(withdrawing, ", ")
+	if set != h.brakeSaid {
+		h.say("WITHDRAWAL BRAKED: this pass would withdraw %d of the %d consumer(s) this provider holds (%s), "+
+			"more than %d at once or %.0f%% of them. Nothing is withdrawn; each is announced as %s (%s), and one "+
+			"is let go every %s while the mesh goes on not asking for them (novox/hq ADR 0227 rule 4)",
+			len(withdrawing), len(h.applied), set, h.WithdrawAtOnce, h.WithdrawFraction*100, EventFailing,
+			ClassWithdrawalBraked, h.ReleaseEvery)
+		h.brakeSaid = set
+	}
+	if len(h.braked) == 0 {
+		// The release clock starts with the brake, not at the last release of an earlier one.
+		h.releasedAt = now
+	}
+	for _, as := range withdrawing {
+		if _, kept := h.braked[as]; !kept {
+			h.braked[as] = now
+		}
+		text := fmt.Sprintf("the mesh no longer asks for it, and the pass that would withdraw it would withdraw %d "+
+			"consumers at once: kept until released (%s)", len(withdrawing), set)
+		h.failed(as, h.applied[as].node, ClassWithdrawalBraked, text)
+	}
+	if now.Sub(h.releasedAt) < h.ReleaseEvery {
+		return nil
+	}
+	h.releasedAt = now
+	release := withdrawing[0]
+	h.say("%s: released by the withdrawal brake after %s; %d more kept", release,
+		now.Sub(h.braked[release]).Round(time.Second), len(withdrawing)-1)
+	h.recovered(release, "withdrawn")
+	return []string{release}
+}
+
 // failed counts one more failure in a consumer's unbroken run, and announces the run once it has
 // lasted FailingAfter — then again every SayAgainEvery while it lasts.
 func (h *Harness) failed(as, node, class, text string) {