retention: one identity is one record, and the shelf is what says so

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

pkg/retention/base.go

@@ -64,6 +64,12 @@ func (d durable) insert(r Record) error {
 	row.Id = r.ID
 	write(row, r)
 	if err := d.app.Save(row); err != nil {
+		// The identity is unique per org, and the index is what makes it so. Asking
+		// what is retained under it turns the refusal into the one the ledger acts
+		// on, without reading the database's own words for it.
+		if prior, taken := d.retained(r.Org, r.identity); taken == nil && prior.ID != r.ID {
+			return fmt.Errorf("%w: %s", ErrConflict, r.identity)
+		}
 		return fmt.Errorf("%w: retaining %s: %w", ErrStore, r.ID, err)
 	}
 	for _, p := range r.Parties {

pkg/retention/shelf_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/hanzoai/base/core"
 	"github.com/hanzoai/base/tests"
 
@@ -390,8 +391,32 @@ func mustOnly(t *testing.T, l *Ledger) string {
 	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.
+// TestOneIdentityOneRecord: whatever else happens — two retries in flight together,
+// two processes writing to one ledger — a second record cannot take an identity that
+// is already taken. The write is attempted directly on the shelf here, because the
+// ledger's own read-then-write is what this is the backstop for.
+func TestOneIdentityOneRecord(t *testing.T) {
+	l := fresh(t)
+
+	id, err := transact(l, org, "", "tx-1", []string{"subject:p1"}, ago(1), body("tx"))
+	if err != nil {
+		t.Fatalf("transaction: %v", err)
+	}
+
+	twin := mustGet(t, l, id)
+	twin.ID = uuid.NewString()
+	if err := l.shelf.insert(twin); !errors.Is(err, ErrConflict) {
+		t.Fatalf("a second record took the identity %q: err = %v, want ErrConflict", twin.identity, err)
+	}
+	if held := mustLen(t, l); held != 1 {
+		t.Errorf("ledger holds %d records under one identity", held)
+	}
+}
+
+// TestConcurrentRetriesRetainOnce: retries in flight together must each come back
+// with the one record. It cannot force the interleaving it is aiming at, so it is
+// evidence about the ordinary case and not about the race; the guarantee under a
+// race is the identity constraint, which TestOneIdentityOneRecord asks for directly.
 func TestConcurrentRetriesRetainOnce(t *testing.T) {
 	l := fresh(t)
 

pkg/retention/store.go

@@ -25,6 +25,9 @@ import "time"
 //     everything or nothing.
 //   - read and retained report [ErrNotFound] for an absent record, so a caller can
 //     tell "not there" from "could not look".
+//   - insert reports [ErrConflict] when the identity is already taken. Two retries
+//     that both find nothing retained are stopped here whatever else happens, so
+//     one identity is one record even if the shelf is written to from two places.
 //   - Reads return whole records, copied. There is no operation that returns part
 //     of one, because the unit of disposal is the whole record and a shelf that
 //     could hand out a piece could hand out a record with a piece missing.