retention: keep the ledger where a restart cannot reach it

The five-year period is the obligation, so a ledger that starts empty every
time the process starts has kept nothing (AMLR Art. 77). The ledger now
stands on a shelf: it holds the obligations and the shelf holds the records,
in memory for exercising them and in Base collections for keeping them.

pkg/store is the one way a durable store is declared here — a Kind, an
idempotent Ensure that brings an existing collection up to the declaration,
a tenant predicate added by the reader rather than by the caller, and field
names as constants both sides of the write name.

A retry is no longer a second record: a record is identified by what is
unique to its class, and an assessment recurs per decision rather than per
case (AMLR Art. 77(1)(b)).

Co-authored-by: Hanzo Dev <[email protected]>

pkg/api/records_test.go

@@ -82,6 +82,16 @@ func ingest(t *testing.T, h *Handler, body any) *httptest.ResponseRecorder {
 	return rec
 }
 
+// ledgerLen is how many records the handler's ledger holds.
+func ledgerLen(t *testing.T, h *Handler) int {
+	t.Helper()
+	held, err := h.Records.Len()
+	if err != nil {
+		t.Fatalf("ledger: %v", err)
+	}
+	return held
+}
+
 func only(t *testing.T, h *Handler, class retention.Class) retention.Record {
 	t.Helper()
 	var found []retention.Record
@@ -113,8 +123,8 @@ func TestIngestRefusesWithoutKeyMaterial(t *testing.T) {
 	}
 
 	// And nothing was stored on the way to refusing.
-	if h.Records.Len() != 0 {
-		t.Errorf("ledger holds %d records", h.Records.Len())
+	if held, err := h.Records.Len(); err != nil || held != 0 {
+		t.Errorf("ledger holds %d records (%v)", held, err)
 	}
 	if h.Alerts.Len() != 0 {
 		t.Errorf("alert store holds %d transactions", h.Alerts.Len())
@@ -466,8 +476,8 @@ func TestResolveRetainsTheDecision(t *testing.T) {
 	if got := h.Cases.Get(c.ID); got.Status == types.CaseClosed {
 		t.Fatal("a case closed without a retained decision")
 	}
-	if h.Records.Len() != 0 {
-		t.Errorf("an incomplete assessment was retained: %d records", h.Records.Len())
+	if held, err := h.Records.Len(); err != nil || held != 0 {
+		t.Errorf("an incomplete assessment was retained: %d records (%v)", held, err)
 	}
 
 	rec := resolve(map[string]any{
@@ -563,7 +573,7 @@ func TestSandboxReplaysTheEngineOverRetainedHistory(t *testing.T) {
 			t.Fatalf("ingest %v: status %d: %s", notional, rec.Code, rec.Body.String())
 		}
 	}
-	records, opened := h.Records.Len(), h.Cases.Len()
+	records, opened := ledgerLen(t, h), h.Cases.Len()
 
 	e, rec := send(http.MethodPost, "/v1/aml/rules/test", map[string]any{
 		"dsl":       "Tx.Notional >= 9000.0 && Tx.Notional < 10000.0",
@@ -601,8 +611,8 @@ func TestSandboxReplaysTheEngineOverRetainedHistory(t *testing.T) {
 	// A sandbox that can mutate live state is not a sandbox. The ingested 15000
 	// opened a case on the way in; the replay must not open another, nor retain
 	// anything, nor alert.
-	if h.Records.Len() != records {
-		t.Errorf("the replay changed the ledger: %d records, was %d", h.Records.Len(), records)
+	if held := ledgerLen(t, h); held != records {
+		t.Errorf("the replay changed the ledger: %d records, was %d", held, records)
 	}
 	if h.Cases.Len() != opened {
 		t.Errorf("the replay opened %d cases", h.Cases.Len()-opened)
@@ -657,8 +667,8 @@ func TestSandboxOverASampleDoesNotTouchTheRecordPlane(t *testing.T) {
 	if !report.From.IsZero() || !report.To.IsZero() {
 		t.Errorf("a sample claims the window %s..%s", report.From, report.To)
 	}
-	if h.Records.Len() != 0 {
-		t.Errorf("a sample replay wrote %d records", h.Records.Len())
+	if held := ledgerLen(t, h); held != 0 {
+		t.Errorf("a sample replay wrote %d records", held)
 	}
 }
 

pkg/retention/base.go

@@ -184,6 +184,11 @@ func (d durable) each(org string, c Class, visit func(Record) error) error {
 			}
 		}
 		last := rows[len(rows)-1]
+		if last.Id == id {
+			// The page did not move, so the next one would be this one again. A walk
+			// that cannot advance says so rather than never returning.
+			return fmt.Errorf("%w: a walk of %s stopped advancing at %s", ErrStore, org, id)
+		}
 		at, id = last.GetDateTime(fieldOccurred).Time(), last.Id
 	}
 }

pkg/retention/ledger_test.go

@@ -56,6 +56,15 @@ func assess(l *Ledger, o, rel, ref string, parties []string, a Assessment) (stri
 	})
 }
 
+func mustLen(t *testing.T, l *Ledger) int {
+	t.Helper()
+	held, err := l.Len()
+	if err != nil {
+		t.Fatalf("Len: %v", err)
+	}
+	return held
+}
+
 func mustGet(t *testing.T, l *Ledger, id string) Record {
 	t.Helper()
 	r, err := l.Get(PurposeInvestigation, org, id)
@@ -69,7 +78,7 @@ func mustGet(t *testing.T, l *Ledger, id string) Record {
 // relationship, from the occasional transaction, or from the date of refusal.
 // The third is the one implementations miss, so it is asserted like the others.
 func TestThreeTriggers(t *testing.T) {
-	l := New()
+	l := fresh(t)
 
 	// From the occasional transaction.
 	occasional, err := transact(l, org, "", "tx-occasional", []string{"subject:p1"}, ago(4), body("occ"))
@@ -123,7 +132,7 @@ func TestThreeTriggers(t *testing.T) {
 // TestRefusalNeedsAReason: a refusal with no reason answers nothing about why
 // the firm refrained, which is the point of retaining it.
 func TestRefusalNeedsAReason(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	if _, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(1), ""); !errors.Is(err, ErrReason) {
 		t.Fatalf("err = %v, want ErrReason", err)
 	}
@@ -132,7 +141,7 @@ func TestRefusalNeedsAReason(t *testing.T) {
 // TestTheClockIsNotTheCallersToSet: a caller that can name its own expiry can
 // name one that never arrives, so Retain refuses a record that brings a clock.
 func TestTheClockIsNotTheCallersToSet(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	_, err := l.Retain(Record{
 		Org: org, Class: ClassRefusal, Trigger: TriggerRefusal, Ref: "tx-1",
 		Parties: []string{"subject:p1"}, Occurred: ago(6), Reason: "refused",
@@ -149,7 +158,7 @@ func TestTheClockIsNotTheCallersToSet(t *testing.T) {
 // TestTriggerAndRelationshipMustAgree: a trigger is a statement about where the
 // period runs from, so it cannot contradict where the record sits.
 func TestTriggerAndRelationshipMustAgree(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	rel, err := open(l, org, "payments", []string{"subject:p1"}, ago(3))
 	if err != nil {
 		t.Fatalf("open: %v", err)
@@ -187,7 +196,7 @@ func TestTriggerAndRelationshipMustAgree(t *testing.T) {
 // its expiry when it is written, because the period runs from the end of the
 // relationship. Closing the relationship starts every one of those clocks.
 func TestClockCascadesOnClose(t *testing.T) {
-	l := New()
+	l := fresh(t)
 
 	rel, err := open(l, org, "brokerage", []string{"subject:p1"}, ago(9))
 	if err != nil {
@@ -239,7 +248,7 @@ func TestClockCascadesOnClose(t *testing.T) {
 // TestCloseRefusals: a relationship closes once, in its own org, and not before
 // it opened.
 func TestCloseRefusals(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	rel, err := open(l, org, "payments", []string{"subject:p1"}, ago(3))
 	if err != nil {
 		t.Fatalf("open: %v", err)
@@ -266,7 +275,7 @@ func TestCloseRefusals(t *testing.T) {
 // TestDisposeDestroysExpiredAndProvesIt: the whole record and every index entry
 // that referenced it, and nothing that is still within its period.
 func TestDisposeDestroysExpiredAndProvesIt(t *testing.T) {
-	l := New()
+	l := fresh(t)
 
 	expired, err := refuse(l, org, "tx-old", []string{"subject:gone", "name:gone"}, ago(6), "refused")
 	if err != nil {
@@ -326,7 +335,7 @@ func TestDisposeDestroysExpiredAndProvesIt(t *testing.T) {
 // TestDisposeCascadeLeavesNoDanglingIndex: closing a relationship expires it and
 // everything inside it together, and the relationship index goes with them.
 func TestDisposeCascadeLeavesNoDanglingIndex(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	rel, err := open(l, org, "payments", []string{"subject:p1"}, ago(12))
 	if err != nil {
 		t.Fatalf("open: %v", err)
@@ -345,18 +354,32 @@ func TestDisposeCascadeLeavesNoDanglingIndex(t *testing.T) {
 	if len(d.Disposed) != 2 {
 		t.Fatalf("disposed %d records, want 2", len(d.Disposed))
 	}
-	if l.Len() != 0 {
-		t.Errorf("ledger holds %d records, want 0", l.Len())
+	if held := mustLen(t, l); held != 0 {
+		t.Errorf("ledger holds %d records, want 0", held)
 	}
-	if len(l.parties) != 0 || len(l.inside) != 0 {
-		t.Errorf("indexes left behind: %d party keys, %d relationships", len(l.parties), len(l.inside))
+
+	// Nothing is findable by either index. The party index is asked through the
+	// answer it feeds, and the relationship index through the cascade it feeds.
+	answer, err := l.Lookback(PurposeDisclosure, org, "subject:p1", time.Now().UTC())
+	if err != nil {
+		t.Fatalf("Lookback: %v", err)
+	}
+	if answer.Examined != 0 {
+		t.Errorf("the party index still holds %d entries", answer.Examined)
+	}
+	retained, err := l.shelf.inside(org, rel)
+	if err != nil {
+		t.Fatalf("inside: %v", err)
+	}
+	if len(retained) != 0 {
+		t.Errorf("the destroyed relationship still indexes %d records", len(retained))
 	}
 }
 
 // TestDisposeRefusesTheFuture: a disposal run cannot be talked into early
 // destruction by a date the clock has not reached.
 func TestDisposeRefusesTheFuture(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	id, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(1), "refused")
 	if err != nil {
 		t.Fatalf("refusal: %v", err)
@@ -375,11 +398,11 @@ func TestDisposeRefusesTheFuture(t *testing.T) {
 // pointing at a record that is not there — because that is the shape of the bug a
 // disposal job hides when it counts only its own delete statements.
 func TestDisposeReportsNoSuccessWhenProofFails(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	if _, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(1), "refused"); err != nil {
 		t.Fatalf("refusal: %v", err)
 	}
-	l.parties[partyKey(org, "subject:ghost")] = []string{"a-record-that-is-not-there"}
+	poison(t, l, "subject:ghost", "a-record-that-is-not-there")
 
 	d, err := l.Dispose(time.Now().UTC())
 	if !errors.Is(err, ErrDisposal) {
@@ -393,7 +416,7 @@ func TestDisposeReportsNoSuccessWhenProofFails(t *testing.T) {
 // TestLookbackAnswersArticle78: is or was a relationship maintained with this
 // party in the prior five years, and what was its nature.
 func TestLookbackAnswersArticle78(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	now := time.Now().UTC()
 
 	// Ended inside the window.
@@ -469,7 +492,7 @@ func TestLookbackAnswersArticle78(t *testing.T) {
 // TestLookbackIgnoresRecordsThatAreNotRelationships: Art. 78 asks about business
 // relationships, and a transaction with a party is not one.
 func TestLookbackIgnoresRecordsThatAreNotRelationships(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	if _, err := transact(l, org, "", "tx-1", []string{"name:one off"}, ago(1), body("x")); err != nil {
 		t.Fatalf("transaction: %v", err)
 	}
@@ -490,7 +513,7 @@ func TestLookbackIgnoresRecordsThatAreNotRelationships(t *testing.T) {
 // available, so the work a lookback does is proportional to one party's records
 // and not to the ledger. Examined is that proof and it stays flat.
 func TestLookbackIsIndexedNotScanned(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	now := time.Now().UTC()
 
 	if _, err := open(l, org, "payments", []string{"name:target"}, ago(3)); err != nil {
@@ -515,7 +538,7 @@ func TestLookbackIsIndexedNotScanned(t *testing.T) {
 	}
 	if after.Examined != before.Examined || after.Examined != 1 {
 		t.Fatalf("examined went from %d to %d as the ledger grew to %d records",
-			before.Examined, after.Examined, l.Len())
+			before.Examined, after.Examined, mustLen(t, l))
 	}
 	if !after.Maintained {
 		t.Error("answer changed as unrelated records were added")
@@ -524,7 +547,7 @@ func TestLookbackIsIndexedNotScanned(t *testing.T) {
 
 // TestOrgBoundary: no read crosses it, in either direction, on any surface.
 func TestOrgBoundary(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	const other = "beta"
 
 	mine, err := open(l, org, "payments", []string{"name:shared party"}, ago(2))
@@ -562,7 +585,7 @@ func TestOrgBoundary(t *testing.T) {
 // TestReadsRefuseOtherPurposes: retained personal data is for preventing money
 // laundering and terrorist financing, so a commercial purpose gets no data.
 func TestReadsRefuseOtherPurposes(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	id, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(1), "refused")
 	if err != nil {
 		t.Fatalf("refusal: %v", err)
@@ -585,7 +608,7 @@ func TestReadsRefuseOtherPurposes(t *testing.T) {
 // reader that alters what it was given has not altered the retained record.
 // Redaction is prohibited, and here it is also not reachable.
 func TestRecordsAreNotRedactableThroughAReader(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	id, err := assess(l, org, "", "case-1", []string{"subject:p1"}, Assessment{
 		Considered: []string{"velocity", "profile"},
 		Result:     NotReported,
@@ -632,7 +655,7 @@ func TestRecordsAreNotRedactableThroughAReader(t *testing.T) {
 // TestExtendRefusals: an extension is a decision somebody made about one case,
 // within five further years, once.
 func TestExtendRefusals(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	id, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(1), "refused")
 	if err != nil {
 		t.Fatalf("refusal: %v", err)
@@ -677,7 +700,7 @@ func TestExtendRefusals(t *testing.T) {
 // TestExtensionSurvivesDisposal: an extended record is not swept up on the date
 // it would otherwise have expired.
 func TestExtensionSurvivesDisposal(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	id, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(6), "refused")
 	if err != nil {
 		t.Fatalf("refusal: %v", err)
@@ -701,7 +724,7 @@ func TestExtensionSurvivesDisposal(t *testing.T) {
 // TestWriteRefusals: what the ledger will not accept, because a record that
 // cannot answer for itself is not worth retaining.
 func TestWriteRefusals(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	rel, err := open(l, org, "payments", []string{"subject:p1"}, ago(1))
 	if err != nil {
 		t.Fatalf("open: %v", err)
@@ -755,7 +778,7 @@ func TestWriteRefusals(t *testing.T) {
 
 // TestEachIsOrderedOldestFirst: a file produced from the ledger is reproducible.
 func TestEachIsOrderedOldestFirst(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	for i, years := range []int{1, 4, 2, 3} {
 		if _, err := refuse(l, org, fmt.Sprint("tx-", i), []string{"subject:p1"}, ago(years), "refused"); err != nil {
 			t.Fatalf("refusal: %v", err)
@@ -782,7 +805,7 @@ func TestEachIsOrderedOldestFirst(t *testing.T) {
 // TestEachFiltersByClass: the class filter is a filter, and an empty one means
 // every class.
 func TestEachFiltersByClass(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	if _, err := refuse(l, org, "tx-1", []string{"subject:p1"}, ago(1), "refused"); err != nil {
 		t.Fatalf("refusal: %v", err)
 	}
@@ -813,7 +836,7 @@ func TestEachFiltersByClass(t *testing.T) {
 
 // TestEachStopsOnError: a caller can abandon the walk, and the error is its own.
 func TestEachStopsOnError(t *testing.T) {
-	l := New()
+	l := fresh(t)
 	for i := range 3 {
 		if _, err := refuse(l, org, fmt.Sprint("tx-", i), []string{"subject:p1"}, ago(1), "refused"); err != nil {
 			t.Fatalf("refusal: %v", err)

pkg/retention/shelf_test.go

@@ -0,0 +1,523 @@
+package retention
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hanzoai/base/core"
+	"github.com/hanzoai/base/tests"
+
+	_ "github.com/hanzoai/base/migrations"
+)
+
+// The suite runs once per shelf. An invariant that holds only in memory is not an
+// invariant of this package: the durable shelf is what an instance serves from, and
+// it is the one where a filter, an index or a forgotten column can quietly turn a
+// held record into no result. So every test below and in ledger_test.go is asked of
+// both, and each one starts from an empty ledger.
+var run struct {
+	name  string
+	fresh func(t *testing.T) *Ledger
+}
+
+func TestMain(m *testing.M) {
+	for _, each := range []struct {
+		name  string
+		fresh func(t *testing.T) *Ledger
+	}{
+		{name: "memory", fresh: func(*testing.T) *Ledger { return New() }},
+		{name: "base", fresh: func(t *testing.T) *Ledger { return NewBase(app(t)) }},
+	} {
+		run = each
+		if code := m.Run(); code != 0 {
+			os.Exit(code)
+		}
+	}
+	os.Exit(0)
+}
+
+// fresh is an empty ledger on the shelf this run is exercising. It names the shelf
+// in the test log so that a failure says which one failed.
+func fresh(t *testing.T) *Ledger {
+	t.Helper()
+	t.Logf("shelf: %s", run.name)
+	return run.fresh(t)
+}
+
+// app is a Base app with the retention collections created.
+func app(t *testing.T) core.App {
+	t.Helper()
+	built, err := tests.NewTestApp()
+	if err != nil {
+		t.Fatalf("test app: %v", err)
+	}
+	t.Cleanup(built.Cleanup)
+	if err := Ensure(built); err != nil {
+		t.Fatalf("Ensure: %v", err)
+	}
+	return built
+}
+
+// poison puts an entry in the party index that names a record which is not there —
+// the state a disposal run leaves behind if it destroys a record and not the index
+// entries that found it. It reaches around the ledger on purpose: the ledger has no
+// operation that produces this state, and the point is what happens when something
+// else has.
+func poison(t *testing.T, l *Ledger, party, id string) {
+	t.Helper()
+	switch s := l.shelf.(type) {
+	case *memory:
+		s.held.party[key(org, party)] = []string{id}
+	case durable:
+		entry, err := parties.New(s.app, org)
+		if err != nil {
+			t.Fatalf("party index: %v", err)
+		}
+		entry.Set(fieldParty, party)
+		entry.Set(fieldRecord, id)
+		if err := s.app.Save(entry); err != nil {
+			t.Fatalf("party index: %v", err)
+		}
+	default:
+		t.Fatalf("no way to corrupt a %T", s)
+	}
+}
+
+// boot opens a Base app on a directory, creating the databases if they are not
+// there. Two of them over one directory, in sequence, is a restart.
+func boot(t *testing.T, dir string) *core.BaseApp {
+	t.Helper()
+	opened := core.NewBaseApp(core.BaseAppConfig{DataDir: dir, EncryptionEnv: "hz_test_env"})
+	if err := opened.Bootstrap(); err != nil {
+		t.Fatalf("bootstrap %s: %v", dir, err)
+	}
+	if err := opened.RunAllMigrations(); err != nil {
+		t.Fatalf("migrate %s: %v", dir, err)
+	}
+	return opened
+}
+
+// TestRecordsSurviveARestart is the whole point of a durable shelf. The retention
+// period is five years (AMLR Art. 77(3)); a ledger that starts empty every time the
+// process starts has kept nothing, and the breach is of the record-keeping
+// obligation itself rather than of anything a customer did.
+func TestRecordsSurviveARestart(t *testing.T) {
+	if run.name != "base" {
+		t.Skip("a memory shelf cannot survive a restart, which is why it is not what an instance serves from")
+	}
+
+	dir := filepath.Join(t.TempDir(), "data")
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+
+	first := boot(t, dir)
+	if err := Ensure(first); err != nil {
+		t.Fatalf("Ensure: %v", err)
+	}
+	before := NewBase(first)
+
+	rel, err := open(before, org, "payments", []string{"name:ivan petrov"}, ago(9))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	tx, err := transact(before, org, rel, "tx-1", []string{"name:ivan petrov"}, ago(8), body("kept"))
+	if err != nil {
+		t.Fatalf("transaction: %v", err)
+	}
+	if _, err := before.Close(org, rel, ago(4)); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if err := first.ResetBootstrapState(); err != nil {
+		t.Fatalf("shutting down: %v", err)
+	}
+
+	second := boot(t, dir)
+	t.Cleanup(func() { _ = second.ResetBootstrapState() })
+	// Ensure runs on every start, and the second run must not disturb what the
+	// first one created.
+	if err := Ensure(second); err != nil {
+		t.Fatalf("Ensure after restart: %v", err)
+	}
+	after := NewBase(second)
+
+	held, err := after.Len()
+	if err != nil {
+		t.Fatalf("Len: %v", err)
+	}
+	if held != 2 {
+		t.Fatalf("ledger holds %d records after a restart, want 2", held)
+	}
+
+	kept, err := after.Get(PurposeInvestigation, org, tx)
+	if err != nil {
+		t.Fatalf("the transaction did not survive the restart: %v", err)
+	}
+	if string(kept.Body) != string(body("kept")) {
+		t.Errorf("body = %q", kept.Body)
+	}
+	// The cascaded clock survived too, not just the record.
+	if want := ago(4).AddDate(Period, 0, 0); !kept.Expiry().Equal(want) {
+		t.Errorf("expiry = %s, want %s", kept.Expiry(), want)
+	}
+
+	// And the index the Art. 78 answer comes from.
+	answer, err := after.Lookback(PurposeDisclosure, org, "name:ivan petrov", time.Now().UTC())
+	if err != nil {
+		t.Fatalf("Lookback: %v", err)
+	}
+	if !answer.Maintained || answer.Current {
+		t.Errorf("answer after restart = %+v, want maintained and not current", answer)
+	}
+	if answer.Examined != 2 {
+		t.Errorf("examined = %d, want 2", answer.Examined)
+	}
+}
+
+// TestEveryFieldSurvivesTheRoundTrip: a column the writer forgets to set does not
+// fail, it reads back empty — and an empty field in a retained record is a
+// transaction that can no longer be reconstructed. So a record with every field
+// populated is written, read back and compared whole.
+func TestEveryFieldSurvivesTheRoundTrip(t *testing.T) {
+	l := fresh(t)
+
+	rel, err := open(l, org, "payments", []string{"name:ivan petrov"}, ago(9))
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	if _, err := l.Close(org, rel, ago(4)); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	decided := ago(3).Add(90 * time.Minute)
+	full := Record{
+		Org:          org,
+		Class:        ClassTransaction,
+		Trigger:      TriggerRelationshipEnd,
+		Relationship: rel,
+		Ref:          "tx-full",
+		Nature:       "payments",
+		Reason:       "kept for the round trip",
+		Parties:      []string{"subject:p1", "name:ivan petrov"},
+		Occurred:     ago(3),
+		Body:         []byte{0x00, 0x01, 0xff, 0xfe, '{', '}'},
+		Assessment: &Assessment{
+			Alerts:     []string{"alert-1", "alert-2"},
+			Case:       "case-1",
+			Considered: []string{"the counterparty", "the pattern"},
+			Result:     NotReported,
+			Rationale:  "explained by the customer's business",
+			By:         "analyst-1",
+			At:         decided,
+		},
+	}
+	id, err := l.Retain(full)
+	if err != nil {
+		t.Fatalf("Retain: %v", err)
+	}
+	if err := l.Extend(org, id, 30*24*time.Hour, "an authority has asked", "officer-1"); err != nil {
+		t.Fatalf("Extend: %v", err)
+	}
+
+	got := mustGet(t, l, id)
+	want := full
+	want.ID = id
+	want.Start = ago(4).UTC()
+	want.Occurred = full.Occurred.UTC()
+	want.Written = got.Written
+	want.Extended = got.Extended
+	want.identity = got.identity
+
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("round trip lost or changed a field:\n got %+v\nwant %+v", got, want)
+	}
+	if got.Extended == nil || got.Extended.Reason != "an authority has asked" || got.Extended.Who != "officer-1" {
+		t.Errorf("extension = %+v", got.Extended)
+	}
+	if got.Written.IsZero() {
+		t.Error("the write time was not kept")
+	}
+	if got.identity == "" {
+		t.Error("the identity was not kept, so a retry would write a second record")
+	}
+}
+
+// TestRetryDoesNotRetainTwice: a client that resends because it never saw the first
+// response must not put a second record of one transaction in the ledger. The
+// second call returns the id the first one wrote.
+func TestRetryDoesNotRetainTwice(t *testing.T) {
+	l := fresh(t)
+
+	first, err := transact(l, org, "", "tx-1", []string{"subject:p1"}, ago(1), body("tx"))
+	if err != nil {
+		t.Fatalf("transaction: %v", err)
+	}
+	again, err := transact(l, org, "", "tx-1", []string{"subject:p1"}, ago(1), body("tx"))
+	if err != nil {
+		t.Fatalf("retry: %v", err)
+	}
+	if again != first {
+		t.Errorf("retry returned %s, want the record it already wrote %s", again, first)
+	}
+
+	held, err := l.Len()
+	if err != nil {
+		t.Fatalf("Len: %v", err)
+	}
+	if held != 1 {
+		t.Fatalf("ledger holds %d records after one transaction and a retry", held)
+	}
+
+	// The party index must not have gained a second entry either, or the same
+	// relationship would be counted twice in an Art. 78 answer.
+	answer, err := l.Lookback(PurposeDisclosure, org, "subject:p1", time.Now().UTC())
+	if err != nil {
+		t.Fatalf("Lookback: %v", err)
+	}
+	if answer.Examined != 1 {
+		t.Errorf("party index holds %d entries for one record", answer.Examined)
+	}
+
+	// A refusal is recognised the same way: one refused transaction, one record.
+	refused, err := refuse(l, org, "tx-2", []string{"subject:p2"}, ago(1), "identification impossible")
+	if err != nil {
+		t.Fatalf("refusal: %v", err)
+	}
+	repeated, err := refuse(l, org, "tx-2", []string{"subject:p2"}, ago(1), "identification impossible")
+	if err != nil {
+		t.Fatalf("refusal retry: %v", err)
+	}
+	if repeated != refused {
+		t.Errorf("refusal retry returned %s, want %s", repeated, refused)
+	}
+}
+
+// TestAssessmentRecursPerCase: an assessment is not unique per case. The same case
+// is assessed again whenever new information arrives and each decision is retained
+// (AMLR Art. 77(1)(b)), so two decisions are two records — while a retry of one
+// decision is still one.
+func TestAssessmentRecursPerCase(t *testing.T) {
+	l := fresh(t)
+
+	monday := Assessment{
+		Considered: []string{"the first alert"},
+		Result:     NotReported,
+		Rationale:  "consistent with the stated business",
+		By:         "analyst-1",
+		At:         ago(2),
+	}
+	tuesday := monday
+	tuesday.Considered = []string{"the first alert", "a second alert"}
+	tuesday.Result = Reported
+	tuesday.Rationale = "the pattern continued"
+	tuesday.At = ago(2).Add(24 * time.Hour)
+
+	one, err := assess(l, org, "", "case-1", []string{"subject:p1"}, monday)
+	if err != nil {
+		t.Fatalf("first assessment: %v", err)
+	}
+	two, err := assess(l, org, "", "case-1", []string{"subject:p1"}, tuesday)
+	if err != nil {
+		t.Fatalf("second assessment: %v", err)
+	}
+	if one == two {
+		t.Fatal("the second decision on the case replaced the first")
+	}
+
+	repeated, err := assess(l, org, "", "case-1", []string{"subject:p1"}, tuesday)
+	if err != nil {
+		t.Fatalf("retry of the second assessment: %v", err)
+	}
+	if repeated != two {
+		t.Errorf("retry wrote %s, want the decision already retained %s", repeated, two)
+	}
+
+	held, err := l.Len()
+	if err != nil {
+		t.Fatalf("Len: %v", err)
+	}
+	if held != 2 {
+		t.Fatalf("ledger holds %d assessments, want the two decisions", held)
+	}
+}
+
+// TestTwoFactsCannotShareOneIdentity: a second record under the same reference that
+// says something different is not a retry. The ledger is append-only, so it will
+// neither rewrite what it holds nor quietly discard what the caller sent — it
+// refuses and lets the caller find out.
+func TestTwoFactsCannotShareOneIdentity(t *testing.T) {
+	l := fresh(t)
+
+	if _, err := transact(l, org, "", "tx-1", []string{"subject:p1"}, ago(1), body("as submitted")); err != nil {
+		t.Fatalf("transaction: %v", err)
+	}
+	if _, err := transact(l, org, "", "tx-1", []string{"subject:p1"}, ago(1), body("something else")); !errors.Is(err, ErrConflict) {
+		t.Fatalf("err = %v, want ErrConflict", err)
+	}
+
+	held, err := l.Len()
+	if err != nil {
+		t.Fatalf("Len: %v", err)
+	}
+	if held != 1 {
+		t.Errorf("ledger holds %d records, want the first one only", held)
+	}
+	kept := mustGet(t, l, mustOnly(t, l))
+	if string(kept.Body) != string(body("as submitted")) {
+		t.Errorf("the retained record was rewritten: %q", kept.Body)
+	}
+}
+
+// mustOnly is the id of the ledger's single record.
+func mustOnly(t *testing.T, l *Ledger) string {
+	t.Helper()
+	var ids []string
+	if err := l.Each(PurposeMonitoring, org, "", func(r Record) error {
+		ids = append(ids, r.ID)
+		return nil
+	}); err != nil {
+		t.Fatalf("Each: %v", err)
+	}
+	if len(ids) != 1 {
+		t.Fatalf("ledger holds %d records, want 1", len(ids))
+	}
+	return ids[0]
+}
+
+// TestConcurrentRetriesRetainOnce: two retries in flight together both find nothing
+// retained before either writes, unless the read and the write are one transaction.
+func TestConcurrentRetriesRetainOnce(t *testing.T) {
+	l := fresh(t)
+
+	const tries = 4
+	ids := make([]string, tries)
+	errs := make([]error, tries)
+	var wg sync.WaitGroup
+	for i := range tries {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ids[i], errs[i] = transact(l, org, "", "tx-1", []string{"subject:p1"}, ago(1), body("tx"))
+		}()
+	}
+	wg.Wait()
+
+	for i, err := range errs {
+		if err != nil {
+			t.Fatalf("retry %d: %v", i, err)
+		}
+		if ids[i] != ids[0] {
+			t.Errorf("retry %d retained %s, retry 0 retained %s", i, ids[i], ids[0])
+		}
+	}
+	held, err := l.Len()
+	if err != nil {
+		t.Fatalf("Len: %v", err)
+	}
+	if held != 1 {
+		t.Errorf("%d concurrent retries retained %d records", tries, held)
+	}
+}
+
+// TestWalkCrossesPages: a walk reads a page at a time, and a ledger holds more than
+// one page. A cursor that does not advance either repeats a page forever or, worse,
+// stops early — and a walk that stops early is a file produced for an authority with
+// records missing from it and nothing to say so.
+func TestWalkCrossesPages(t *testing.T) {
+	l := fresh(t)
+
+	const held = batch + batch/2
+	first := ago(2)
+	for i := range held {
+		ref := fmt.Sprintf("tx-%04d", i)
+		if _, err := transact(l, org, "", ref, []string{"subject:p1"}, first.Add(time.Duration(i)*time.Minute), body(ref)); err != nil {
+			t.Fatalf("%s: %v", ref, err)
+		}
+	}
+
+	seen := make(map[string]bool, held)
+	previous := time.Time{}
+	order := 0
+	if err := l.Each(PurposeMonitoring, org, ClassTransaction, func(r Record) error {
+		if seen[r.ID] {
+			return fmt.Errorf("record %s visited twice", r.ID)
+		}
+		seen[r.ID] = true
+		if r.Occurred.Before(previous) {
+			return fmt.Errorf("record %d occurred %s, after the one before it at %s", order, r.Occurred, previous)
+		}
+		previous = r.Occurred
+		order++
+		return nil
+	}); err != nil {
+		t.Fatalf("Each: %v", err)
+	}
+	if len(seen) != held {
+		t.Errorf("a walk of %d records visited %d", held, len(seen))
+	}
+}
+
+// TestDisposalCrossesBatches: destruction is done in bounded pieces, and a ledger
+// with more expired records than one piece must still end up with none.
+func TestDisposalCrossesBatches(t *testing.T) {
+	l := fresh(t)
+
+	const doomed = batch + 5
+	for i := range doomed {
+		ref := fmt.Sprintf("tx-%04d", i)
+		if _, err := refuse(l, org, ref, []string{"subject:" + ref}, ago(6), "refused"); err != nil {
+			t.Fatalf("%s: %v", ref, err)
+		}
+	}
+	live, err := refuse(l, org, "tx-recent", []string{"subject:recent"}, ago(1), "refused")
+	if err != nil {
+		t.Fatalf("recent refusal: %v", err)
+	}
+
+	d, err := l.Dispose(time.Now().UTC())
+	if err != nil {
+		t.Fatalf("Dispose: %v", err)
+	}
+	if len(d.Disposed) != doomed {
+		t.Errorf("disposed %d records, want %d", len(d.Disposed), doomed)
+	}
+	if _, err := l.Get(PurposeInvestigation, org, live); err != nil {
+		t.Errorf("a record within its period was destroyed: %v", err)
+	}
+	if held, err := l.Len(); err != nil || held != 1 {
+		t.Errorf("ledger holds %d records after disposal (%v), want the one within its period", held, err)
+	}
+}
+
+// TestEnsureIsIdempotent: it runs on every start, so the second run must find
+// everything already there and change nothing.
+func TestEnsureIsIdempotent(t *testing.T) {
+	if run.name != "base" {
+		t.Skip("Ensure is the durable shelf's")
+	}
+	built := app(t)
+	before, err := built.FindCollectionByNameOrId(records.Name)
+	if err != nil {
+		t.Fatalf("collection: %v", err)
+	}
+	indexes := len(before.Indexes)
+	fields := len(before.Fields)
+
+	if err := Ensure(built); err != nil {
+		t.Fatalf("second Ensure: %v", err)
+	}
+	after, err := built.FindCollectionByNameOrId(records.Name)
+	if err != nil {
+		t.Fatalf("collection: %v", err)
+	}
+	if len(after.Indexes) != indexes || len(after.Fields) != fields {
+		t.Errorf("a second Ensure changed the collection: %d/%d fields, %d/%d indexes",
+			len(after.Fields), fields, len(after.Indexes), indexes)
+	}
+}

pkg/store/store.go

@@ -34,8 +34,10 @@ package store
 import (
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/hanzoai/base/core"
+	"github.com/hanzoai/base/tools/types"
 	"github.com/hanzoai/dbx"
 )
 
@@ -153,13 +155,39 @@ func (k Kind) Count(app core.App) (int, error) {
 }
 
 func (k Kind) records(app core.App, filter, sort string, limit int, params dbx.Params) ([]*core.Record, error) {
-	out, err := app.FindRecordsByFilter(k.Name, filter, sort, limit, 0, params)
+	out, err := app.FindRecordsByFilter(k.Name, filter, sort, limit, 0, bind(params))
 	if err != nil {
 		return nil, fmt.Errorf("store: %s: %w", k.Name, err)
 	}
 	return out, nil
 }
 
+// bind puts the parameters in the form the store compares values in.
+//
+// A moment is the one that has to be converted. Dates are stored in a single fixed
+// layout and compared as text; a Go time bound straight into the filter arrives in
+// a different layout and compares as neither greater nor less than what it means.
+// Nothing errors — the query simply answers with the wrong rows, which for a walk
+// that pages by the last record seen is a page that never ends, and for an expiry
+// check is a record destroyed a year early or never at all. Converting here rather
+// than at each call site is what makes that unforgettable.
+func bind(params dbx.Params) dbx.Params {
+	for name, value := range params {
+		moment, ok := value.(time.Time)
+		if !ok {
+			continue
+		}
+		converted, err := types.ParseDateTime(moment)
+		if err != nil {
+			// ParseDateTime accepts every time.Time; a value it rejects would be a
+			// filter comparing against something that is not a moment at all.
+			continue
+		}
+		params[name] = converted
+	}
+	return params
+}
+
 // name is an index's full name, which is unique across collections because an
 // index name is a schema-wide identifier.
 func (k Kind) name(i Index) string { return "idx_" + k.Name + "_" + i.Name }

pkg/store/store_test.go

@@ -0,0 +1,203 @@
+package store
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hanzoai/base/core"
+	"github.com/hanzoai/base/tests"
+	"github.com/hanzoai/dbx"
+
+	_ "github.com/hanzoai/base/migrations"
+)
+
+func app(t *testing.T) core.App {
+	t.Helper()
+	built, err := tests.NewTestApp()
+	if err != nil {
+		t.Fatalf("test app: %v", err)
+	}
+	t.Cleanup(built.Cleanup)
+	return built
+}
+
+// thing is a Kind to exercise the convention with. It deliberately does not declare
+// Org: Ensure adds it, and a Kind that could omit it would be a collection whose
+// records belong to nobody.
+var thing = Kind{
+	Name: "store_thing",
+	Fields: []core.Field{
+		&core.TextField{Name: "label", Required: true},
+		&core.DateField{Name: "at", Required: true},
+	},
+	Indexes: []Index{
+		{Name: "label", Fields: []string{Org, "label"}},
+	},
+}
+
+func TestEnsureAddsTheTenantAndTheIndexes(t *testing.T) {
+	built := app(t)
+	if err := thing.Ensure(built); err != nil {
+		t.Fatalf("Ensure: %v", err)
+	}
+
+	collection, err := built.FindCollectionByNameOrId(thing.Name)
+	if err != nil {
+		t.Fatalf("collection: %v", err)
+	}
+	if collection.Fields.GetByName(Org) == nil {
+		t.Error("the collection has no tenant field, so a record could belong to nobody")
+	}
+	if !indexed(collection.Indexes, thing.name(thing.Indexes[0])) {
+		t.Errorf("index missing: %v", collection.Indexes)
+	}
+}
+
+// TestEnsureIsIdempotentAndAdditive: Ensure runs on every start. A second run must
+// change nothing, and a run of a declaration that has grown must add what is new
+// without disturbing the records already there — a field the reader expects and the
+// collection lacks reads as empty, which is indistinguishable from a record that
+// never had it.
+func TestEnsureIsIdempotentAndAdditive(t *testing.T) {
+	built := app(t)
+	if err := thing.Ensure(built); err != nil {
+		t.Fatalf("Ensure: %v", err)
+	}
+
+	row, err := thing.New(built, "acme")
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	row.Set("label", "before")
+	row.Set("at", time.Now().UTC())
+	if err := built.Save(row); err != nil {
+		t.Fatalf("Save: %v", err)
+	}
+
+	if err := thing.Ensure(built); err != nil {
+		t.Fatalf("second Ensure: %v", err)
+	}
+
+	grown := thing
+	grown.Fields = append(append([]core.Field{}, thing.Fields...), &core.TextField{Name: "note"})
+	grown.Indexes = append(append([]Index{}, thing.Indexes...), Index{Name: "at", Fields: []string{Org, "at"}})
+	if err := grown.Ensure(built); err != nil {
+		t.Fatalf("Ensure of a grown declaration: %v", err)
+	}
+
+	collection, err := built.FindCollectionByNameOrId(thing.Name)
+	if err != nil {
+		t.Fatalf("collection: %v", err)
+	}
+	if collection.Fields.GetByName("note") == nil {
+		t.Error("a field added to the declaration did not reach the collection")
+	}
+	if !indexed(collection.Indexes, grown.name(grown.Indexes[1])) {
+		t.Errorf("an index added to the declaration did not reach the collection: %v", collection.Indexes)
+	}
+
+	kept, err := grown.Find(built, "acme", "label = {:label}", "", 0, dbx.Params{"label": "before"})
+	if err != nil {
+		t.Fatalf("Find: %v", err)
+	}
+	if len(kept) != 1 {
+		t.Fatalf("the record written before the collection grew is gone: %d records", len(kept))
+	}
+	if kept[0].GetString("note") != "" {
+		t.Errorf("note = %q, want empty on a record written before the field existed", kept[0].GetString("note"))
+	}
+}
+
+// TestQueryIsScopedToOneOrg: the tenant predicate is added by Find and not by the
+// caller's filter, because a filter that forgets it reads another tenant's records
+// and nothing about the result says so.
+func TestQueryIsScopedToOneOrg(t *testing.T) {
+	built := app(t)
+	if err := thing.Ensure(built); err != nil {
+		t.Fatalf("Ensure: %v", err)
+	}
+	for _, org := range []string{"acme", "beta"} {
+		row, err := thing.New(built, org)
+		if err != nil {
+			t.Fatalf("New: %v", err)
+		}
+		row.Set("label", "shared")
+		row.Set("at", time.Now().UTC())
+		if err := built.Save(row); err != nil {
+			t.Fatalf("Save: %v", err)
+		}
+	}
+
+	mine, err := thing.Find(built, "acme", "label = {:label}", "", 0, dbx.Params{"label": "shared"})
+	if err != nil {
+		t.Fatalf("Find: %v", err)
+	}
+	if len(mine) != 1 || mine[0].GetString(Org) != "acme" {
+		t.Fatalf("a scoped query returned %d records: %v", len(mine), mine)
+	}
+
+	if _, err := thing.Find(built, "", "", "", 0, nil); err == nil {
+		t.Error("an unscoped query was answered")
+	}
+	if _, err := thing.New(built, ""); err == nil {
+		t.Error("a record with no organisation was built")
+	}
+
+	// Across is the deliberate exception, and the only one.
+	every, err := thing.Across(built, "", "", 0, nil)
+	if err != nil {
+		t.Fatalf("Across: %v", err)
+	}
+	if len(every) != 2 {
+		t.Errorf("Across saw %d records, want both orgs", len(every))
+	}
+
+	held, err := thing.Count(built)
+	if err != nil {
+		t.Fatalf("Count: %v", err)
+	}
+	if held != 2 {
+		t.Errorf("Count = %d, want both orgs", held)
+	}
+}
+
+// TestAMomentComparesAsAMoment: a date is stored in one layout and compared as
+// text, so a filter parameter that is a Go time has to be converted before it is
+// bound. Unconverted it does not error — it answers with the wrong rows, and the
+// direction it is wrong in depends on the layout rather than on the data.
+func TestAMomentComparesAsAMoment(t *testing.T) {
+	built := app(t)
+	if err := thing.Ensure(built); err != nil {
+		t.Fatalf("Ensure: %v", err)
+	}
+
+	now := time.Now().UTC()
+	for label, at := range map[string]time.Time{
+		"old":    now.Add(-90 * time.Minute),
+		"recent": now.Add(-30 * time.Minute),
+		"later":  now.Add(30 * time.Minute),
+	} {
+		row, err := thing.New(built, "acme")
+		if err != nil {
+			t.Fatalf("New: %v", err)
+		}
+		row.Set("label", label)
+		row.Set("at", at)
+		if err := built.Save(row); err != nil {
+			t.Fatalf("Save: %v", err)
+		}
+	}
+
+	since := now.Add(-time.Hour)
+	within, err := thing.Find(built, "acme", "at >= {:since} && at <= {:now}", "at", 0, dbx.Params{"since": since, "now": now})
+	if err != nil {
+		t.Fatalf("Find: %v", err)
+	}
+	if len(within) != 1 || within[0].GetString("label") != "recent" {
+		labels := make([]string, 0, len(within))
+		for _, r := range within {
+			labels = append(labels, r.GetString("label"))
+		}
+		t.Fatalf("the window matched %v, want [recent]", labels)
+	}
+}