From 27a1dafe27a61b3ff56d6632bedd8fc734ef8cf3 Mon Sep 17 00:00:00 2001 From: bigboateng Date: Sun, 23 Aug 2026 21:04:02 +0100 Subject: [PATCH 1/3] Persist durable, isolated, idempotently resumable control instances The kernel Store is now an explicit multi-instance persistence contract: atomic Create that cannot overwrite, Load returning the atomic instance record (state plus complete ordered receipt history), and per-instance revision CAS, locking, recovery, and history. Apply reconciles an exact committed retry to its original durable receipt with zero side effects, keeps uncommitted attempts recovery-required, and fails closed on fabricated, misrouted, duplicated, reordered, or truncated history. A reusable InstanceStoreConformance suite proves the law for the integer memory store, the settlement register store, and the reviewer's on-disk file store, with white-box counterexamples rejecting dishonest stores. --- boatstack/cmd/boatstack-reviewer/domain.go | 2 +- .../boatstack-reviewer/instance_store_test.go | 26 + boatstack/cmd/boatstack-reviewer/main.go | 47 +- .../cmd/boatstack-reviewer/reviewer_test.go | 21 +- boatstack/cmd/boatstack-reviewer/seal.go | 2 +- boatstack/cmd/boatstack-reviewer/store.go | 140 ++- .../behavior/kernel_backend_test.go | 18 +- boatstack/kernel/conformance/conformance.go | 10 +- .../kernel/conformance/conformance_test.go | 66 +- .../kernel/conformance/instance_store.go | 1005 +++++++++++++++++ .../conformance/instance_store_law_test.go | 275 +++++ boatstack/kernel/conformance/integer.go | 144 ++- boatstack/kernel/conformance/lifecycle.go | 17 +- .../kernel/conformance/lifecycle_law_test.go | 19 +- .../kernel/conformance/lifecycle_register.go | 2 +- .../kernel/conformance/revisioned_register.go | 131 ++- boatstack/kernel/conformance/settlement.go | 13 +- .../kernel/conformance/settlement_law_test.go | 39 +- boatstack/kernel/instance.go | 98 ++ boatstack/kernel/runtime.go | 153 ++- docs/architecture/kernel.md | 72 +- .../2026-08-23-durable-control-instances.md | 35 + 22 files changed, 2119 insertions(+), 216 deletions(-) create mode 100644 boatstack/cmd/boatstack-reviewer/instance_store_test.go create mode 100644 boatstack/kernel/conformance/instance_store.go create mode 100644 boatstack/kernel/conformance/instance_store_law_test.go create mode 100644 boatstack/kernel/instance.go create mode 100644 release-notes/2026-08-23-durable-control-instances.md diff --git a/boatstack/cmd/boatstack-reviewer/domain.go b/boatstack/cmd/boatstack-reviewer/domain.go index e99098a..ae2da15 100644 --- a/boatstack/cmd/boatstack-reviewer/domain.go +++ b/boatstack/cmd/boatstack-reviewer/domain.go @@ -67,7 +67,7 @@ func (d *reviewDomain) observeValue() (observationValue, error) { return observationValue{}, err } value := observationValue{ - Instance: d.store.initial.InstanceID, + Instance: d.store.instance, BaseRef: d.baseRef, MergeBase: mergeBase, HeadCommit: head, diff --git a/boatstack/cmd/boatstack-reviewer/instance_store_test.go b/boatstack/cmd/boatstack-reviewer/instance_store_test.go new file mode 100644 index 0000000..3b46fe1 --- /dev/null +++ b/boatstack/cmd/boatstack-reviewer/instance_store_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "testing" + + "github.com/operatorstack/boatstack/boatstack/kernel" + "github.com/operatorstack/boatstack/boatstack/kernel/conformance" +) + +// TestFileStoreInstanceConformance proves the reviewer's file store is a +// durable, isolated, idempotently resumable multi-instance kernel Store: +// reopening constructs a fresh handle over the same on-disk substrate, so +// restart reconstruction exercises real file persistence. +func TestFileStoreInstanceConformance(t *testing.T) { + conformance.InstanceStoreConformance{New: func(t testing.TB) conformance.InstanceStoreHarness { + gitDir := t.TempDir() + store := newFileStore(gitDir, "instance-alpha") + return conformance.InstanceStoreHarness{ + Store: store, + Locker: func() kernel.Locker { return directoryLocker{store: store} }, + Reopen: func(testing.TB) kernel.Store { + return newFileStore(gitDir, "instance-alpha") + }, + } + }}.Run(t) +} diff --git a/boatstack/cmd/boatstack-reviewer/main.go b/boatstack/cmd/boatstack-reviewer/main.go index f8d9267..cce6472 100644 --- a/boatstack/cmd/boatstack-reviewer/main.go +++ b/boatstack/cmd/boatstack-reviewer/main.go @@ -111,7 +111,7 @@ func newLoopContext(repoPath, delivery, baseRef string) (*loopContext, error) { return nil, err } } - store := newFileStore(repo.GitDir, instance, program.Identity()) + store := newFileStore(repo.GitDir, instance) domain := &reviewDomain{repo: repo, store: store, policy: policy, baseRef: baseRef} return &loopContext{ repo: repo, @@ -128,7 +128,7 @@ func newLoopContext(repoPath, delivery, baseRef string) (*loopContext, error) { func (c *loopContext) runtime() (kernel.Runtime, error) { return kernel.NewRuntime( c.program, c.domain, c.operator, reviewCapabilities{}, - c.store, directoryLocker{path: c.store.lockPath()}, c.clock, + c.store, directoryLocker{store: c.store}, c.clock, ) } @@ -136,6 +136,19 @@ func (c *loopContext) authority(actor string, capabilities ...kernel.Capability) return localAuthority(actor, c.clock.Now(), capabilities...) } +// ensureProvisioned provisions the command's control instance when it does +// not exist yet. Only the mutating entrypoints call it; read-only commands +// surface the typed not-found result instead of manufacturing state. +func (c *loopContext) ensureProvisioned(runtime kernel.Runtime) error { + if _, err := c.store.Load(context.Background(), c.instance); !kernel.IsInstanceNotFound(err) { + return err + } + if _, err := runtime.Provision(context.Background(), c.instance); err != nil && !kernel.IsInstanceExists(err) { + return err + } + return nil +} + func printJSON(value any) error { encoder := json.NewEncoder(os.Stdout) encoder.SetIndent("", " ") @@ -194,6 +207,9 @@ func commandResolve(arguments []string) error { if err != nil { return err } + if err := loop.ensureProvisioned(runtime); err != nil { + return err + } authority, err := loop.authority(*actor, capabilitySubmit) if err != nil { return err @@ -276,6 +292,9 @@ func commandSubmit(arguments []string) error { if err != nil { return err } + if err := loop.ensureProvisioned(runtime); err != nil { + return err + } authority, err := loop.authority(*actor, capabilitySubmit) if err != nil { return err @@ -300,10 +319,11 @@ func commandSubmit(arguments []string) error { if err != nil { return fmt.Errorf("submission did not commit: %w", err) } - state, err := loop.store.Load(context.Background(), loop.instance) + record, err := loop.store.Load(context.Background(), loop.instance) if err != nil { return err } + state := record.State observed, err := loop.domain.observeValue() if err != nil { return err @@ -346,10 +366,14 @@ func commandStatus(arguments []string) error { if err != nil { return err } - state, err := loop.store.Load(context.Background(), loop.instance) + record, err := loop.store.Load(context.Background(), loop.instance) + if kernel.IsInstanceNotFound(err) { + return fmt.Errorf("%w; run `boatstack-reviewer resolve` to provision it", err) + } if err != nil { return err } + state := record.State observed, err := loop.domain.observeValue() if err != nil { return err @@ -392,10 +416,14 @@ func commandShow(arguments []string) error { if err != nil { return err } - state, err := loop.store.Load(context.Background(), loop.instance) + record, err := loop.store.Load(context.Background(), loop.instance) + if kernel.IsInstanceNotFound(err) { + return fmt.Errorf("%w; run `boatstack-reviewer resolve` to provision it", err) + } if err != nil { return err } + state := record.State journal, err := loop.store.loadJournal() if err != nil { return err @@ -491,7 +519,7 @@ func commandSeal(arguments []string) error { if !report.Verified { return fmt.Errorf("seal refused: the full receipt does not verify: %s", strings.Join(report.Failures, "; ")) } - archive := filepath.Join(loop.store.dir, "sealed-receipt.json") + archive := filepath.Join(loop.store.dir(), "sealed-receipt.json") if err := writeSealedReceipt(archive, receipt); err != nil { return err } @@ -576,6 +604,9 @@ func commandRequested(arguments []string, name, transition string, capability ke if err != nil { return err } + if err := loop.ensureProvisioned(runtime); err != nil { + return err + } authority, err := loop.authority(*actor, capability) if err != nil { return err @@ -615,9 +646,9 @@ func commandReset(arguments []string) error { return err } if !*confirm { - return fmt.Errorf("reset archives %s; pass --confirm to proceed", loop.store.dir) + return fmt.Errorf("reset archives %s; pass --confirm to proceed", loop.store.dir()) } - if _, err := os.Stat(loop.store.dir); os.IsNotExist(err) { + if _, err := os.Stat(loop.store.dir()); os.IsNotExist(err) { return fmt.Errorf("instance %s has no local review state", loop.instance) } archived, err := loop.store.archive(time.Now().UTC().Format("20060102T150405Z")) diff --git a/boatstack/cmd/boatstack-reviewer/reviewer_test.go b/boatstack/cmd/boatstack-reviewer/reviewer_test.go index 50d9840..7c7c918 100644 --- a/boatstack/cmd/boatstack-reviewer/reviewer_test.go +++ b/boatstack/cmd/boatstack-reviewer/reviewer_test.go @@ -97,9 +97,9 @@ func newTestLoop(t *testing.T, scratch *scratchRepo, policy Policy) *loopContext if err != nil { t.Fatal(err) } - store := newFileStore(scratch.repo.GitDir, "feature", program.Identity()) + store := newFileStore(scratch.repo.GitDir, "feature") domain := &reviewDomain{repo: scratch.repo, store: store, policy: policy, baseRef: "main"} - return &loopContext{ + loop := &loopContext{ repo: scratch.repo, policy: policy, program: program, @@ -109,6 +109,14 @@ func newTestLoop(t *testing.T, scratch *scratchRepo, policy Policy) *loopContext instance: "feature", baseRef: "main", } + runtime, err := loop.runtime() + if err != nil { + t.Fatal(err) + } + if err := loop.ensureProvisioned(runtime); err != nil { + t.Fatal(err) + } + return loop } func testPolicy(t *testing.T, scratch *scratchRepo) Policy { @@ -195,11 +203,11 @@ func submit(t *testing.T, loop *loopContext, candidate string) (kernel.Resolutio func mode(t *testing.T, loop *loopContext) string { t.Helper() - state, err := loop.store.Load(context.Background(), loop.instance) + record, err := loop.store.Load(context.Background(), loop.instance) if err != nil { t.Fatal(err) } - return state.Mode + return record.State.Mode } func TestReviewProgramControlLaw(t *testing.T) { @@ -950,10 +958,11 @@ func TestRecoveryClearsInterruptedEffect(t *testing.T) { // Simulate a crash between BeginEffect and CommitTransition: the store // holds an attempt revision with an active recovery state. - state, err := loop.store.Load(context.Background(), loop.instance) + record, err := loop.store.Load(context.Background(), loop.instance) if err != nil { t.Fatal(err) } + state := record.State attempt := state attempt.Revision = state.Revision + 1 attempt.Recovery = &kernel.RecoveryState{ @@ -961,7 +970,7 @@ func TestRecoveryClearsInterruptedEffect(t *testing.T) { TransitionID: transitionRecord, Reason: "simulated crash between effect and commit", } - if err := loop.store.BeginEffect(context.Background(), state.Revision, attempt); err != nil { + if err := loop.store.BeginEffect(context.Background(), loop.instance, state.Revision, attempt); err != nil { t.Fatal(err) } if err := loop.store.stageCandidate([]byte(correctReview()), "half-recorded"); err != nil { diff --git a/boatstack/cmd/boatstack-reviewer/seal.go b/boatstack/cmd/boatstack-reviewer/seal.go index 85f5b4e..a33dae9 100644 --- a/boatstack/cmd/boatstack-reviewer/seal.go +++ b/boatstack/cmd/boatstack-reviewer/seal.go @@ -80,7 +80,7 @@ func (r SealedReceipt) contentFingerprint() (string, error) { // refuses unless the instance is converged and the converged round binds the // exact current reviewed tree. func buildSealedReceipt(repo *gitRepo, store *fileStore, policy Policy, program kernel.Program, baseRef string, now time.Time) (SealedReceipt, error) { - document, err := store.loadDocument() + document, err := store.loadDocument(store.instance) if err != nil { return SealedReceipt{}, err } diff --git a/boatstack/cmd/boatstack-reviewer/store.go b/boatstack/cmd/boatstack-reviewer/store.go index 560afc1..83aea6e 100644 --- a/boatstack/cmd/boatstack-reviewer/store.go +++ b/boatstack/cmd/boatstack-reviewer/store.go @@ -53,33 +53,36 @@ type stagedCandidate struct { ReviewedTree string `json:"reviewed_tree"` } -// fileStore owns one review control instance's durable state under the -// repository's .git directory, so nothing here can enter a commit. +// fileStore owns review control instances' durable state under the +// repository's .git directory, so nothing here can enter a commit. The +// kernel-facing Store methods route by the explicit instance identity in +// every call; the domain-side helpers (journal, staging, rounds, sealing) +// act on the command-bound instance. type fileStore struct { - dir string - initial kernel.ControlState + root string + instance string } -func newFileStore(gitDir, instanceID string, program kernel.ProgramIdentity) *fileStore { - return &fileStore{ - dir: filepath.Join(gitDir, "boatstack-review", instanceID), - initial: kernel.ControlState{ - InstanceID: instanceID, - Program: program, - Mode: modeUnreviewed, - Revision: 1, - }, - } +func newFileStore(gitDir, instanceID string) *fileStore { + return &fileStore{root: filepath.Join(gitDir, "boatstack-review"), instance: instanceID} +} + +func (s *fileStore) instanceDir(instanceID string) string { + return filepath.Join(s.root, instanceID) } -func (s *fileStore) statePath() string { return filepath.Join(s.dir, "store.json") } -func (s *fileStore) journalPath() string { return filepath.Join(s.dir, "journal.json") } -func (s *fileStore) stagingPath() string { return filepath.Join(s.dir, "candidate.json") } +// dir is the command-bound instance's durable directory. +func (s *fileStore) dir() string { return s.instanceDir(s.instance) } + +func (s *fileStore) statePath(instanceID string) string { + return filepath.Join(s.instanceDir(instanceID), "store.json") +} +func (s *fileStore) journalPath() string { return filepath.Join(s.dir(), "journal.json") } +func (s *fileStore) stagingPath() string { return filepath.Join(s.dir(), "candidate.json") } func (s *fileStore) stagingMetaPath() string { - return filepath.Join(s.dir, "candidate-meta.json") + return filepath.Join(s.dir(), "candidate-meta.json") } -func (s *fileStore) roundsDir() string { return filepath.Join(s.dir, "rounds") } -func (s *fileStore) lockPath() string { return filepath.Join(s.dir, "lock") } +func (s *fileStore) roundsDir() string { return filepath.Join(s.dir(), "rounds") } func writeFileAtomic(path string, value []byte) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { @@ -106,54 +109,91 @@ func writeFileAtomic(path string, value []byte) error { return nil } -func (s *fileStore) loadDocument() (storeDocument, error) { - value, err := os.ReadFile(s.statePath()) +func (s *fileStore) loadDocument(instanceID string) (storeDocument, error) { + value, err := os.ReadFile(s.statePath(instanceID)) if os.IsNotExist(err) { - return storeDocument{SchemaVersion: storeSchemaVersion, State: s.initial}, nil + return storeDocument{}, kernel.InstanceNotFoundError{InstanceID: instanceID} } if err != nil { return storeDocument{}, err } var document storeDocument if err := json.Unmarshal(value, &document); err != nil { - return storeDocument{}, fmt.Errorf("review store %s does not decode: %w", s.statePath(), err) + return storeDocument{}, fmt.Errorf("review store %s does not decode: %w", s.statePath(instanceID), err) } if document.SchemaVersion != storeSchemaVersion { - return storeDocument{}, fmt.Errorf("review store %s has unsupported schema version %d", s.statePath(), document.SchemaVersion) + return storeDocument{}, fmt.Errorf("review store %s has unsupported schema version %d", s.statePath(instanceID), document.SchemaVersion) } return document, nil } -func (s *fileStore) saveDocument(document storeDocument) error { +func (s *fileStore) saveDocument(instanceID string, document storeDocument) error { encoded, err := json.MarshalIndent(document, "", " ") if err != nil { return err } - return writeFileAtomic(s.statePath(), append(encoded, '\n')) + return writeFileAtomic(s.statePath(instanceID), append(encoded, '\n')) } -func (s *fileStore) Load(context.Context, string) (kernel.ControlState, error) { - document, err := s.loadDocument() +// Create provisions one control instance atomically and never overwrites an +// existing record: the document is written aside and hard-linked into place, +// so exactly one of any number of concurrent creators wins. +func (s *fileStore) Create(_ context.Context, instanceID string, initial kernel.ControlState) error { + if initial.InstanceID != instanceID { + return fmt.Errorf("initial control state belongs to %q, not %q", initial.InstanceID, instanceID) + } + encoded, err := json.MarshalIndent(storeDocument{SchemaVersion: storeSchemaVersion, State: initial}, "", " ") + if err != nil { + return err + } + path := s.statePath(instanceID) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + temp, err := os.CreateTemp(filepath.Dir(path), ".tmp-create-*") if err != nil { - return kernel.ControlState{}, err + return err } - return document.State, nil + tempPath := temp.Name() + defer os.Remove(tempPath) + if _, err := temp.Write(append(encoded, '\n')); err != nil { + temp.Close() + return err + } + if err := temp.Close(); err != nil { + return err + } + if err := os.Link(tempPath, path); err != nil { + if os.IsExist(err) { + return kernel.InstanceExistsError{InstanceID: instanceID} + } + return err + } + return nil } -func (s *fileStore) BeginEffect(_ context.Context, revision uint64, target kernel.ControlState) error { - document, err := s.loadDocument() +func (s *fileStore) Load(_ context.Context, instanceID string) (kernel.InstanceRecord, error) { + document, err := s.loadDocument(instanceID) + if err != nil { + return kernel.InstanceRecord{}, err + } + return kernel.InstanceRecord{State: document.State, Receipts: document.Receipts}, nil +} + +func (s *fileStore) BeginEffect(_ context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { + document, err := s.loadDocument(instanceID) if err != nil { return err } if document.State.Revision != revision { return fmt.Errorf("stale revision: store has %d, attempt expects %d", document.State.Revision, revision) } - document.State = target - return s.saveDocument(document) + document.State = attempt + return s.saveDocument(instanceID, document) } -func (s *fileStore) CommitTransition(_ context.Context, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { - document, err := s.loadDocument() +func (s *fileStore) CommitTransition(_ context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { + document, err := s.loadDocument(instanceID) if err != nil { return err } @@ -162,7 +202,7 @@ func (s *fileStore) CommitTransition(_ context.Context, revision uint64, target } document.State = target document.Receipts = append(document.Receipts, receipt) - return s.saveDocument(document) + return s.saveDocument(instanceID, document) } func (s *fileStore) loadJournal() (journalDocument, error) { @@ -278,8 +318,8 @@ func (s *fileStore) roundBytes(fingerprint string) ([]byte, error) { // archive moves the whole instance directory aside, preserving receipts and // journal for inspection while releasing the instance identity. func (s *fileStore) archive(timestamp string) (string, error) { - archived := s.dir + "-archived-" + timestamp - if err := os.Rename(s.dir, archived); err != nil { + archived := s.dir() + "-archived-" + timestamp + if err := os.Rename(s.dir(), archived); err != nil { return "", err } return archived, nil @@ -297,22 +337,24 @@ func (s *fileStore) nextGeneration() error { return s.clearStagedCandidate() } -// directoryLocker serializes one control instance with an exclusive lock -// directory. A crash can leave the lock behind; the error names the exact -// path so the operator can remove a stale lock deliberately. -type directoryLocker struct{ path string } +// directoryLocker serializes each control instance independently with an +// exclusive lock directory keyed by the requested instance identity. A crash +// can leave the lock behind; the error names the exact path so the operator +// can remove a stale lock deliberately. +type directoryLocker struct{ store *fileStore } -func (l directoryLocker) Acquire(context.Context, string) (kernel.Lock, error) { - if err := os.MkdirAll(filepath.Dir(l.path), 0o755); err != nil { +func (l directoryLocker) Acquire(_ context.Context, instanceID string) (kernel.Lock, error) { + path := filepath.Join(l.store.instanceDir(instanceID), "lock") + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return nil, err } - if err := os.Mkdir(l.path, 0o755); err != nil { + if err := os.Mkdir(path, 0o755); err != nil { if os.IsExist(err) { - return nil, fmt.Errorf("another boatstack-reviewer invocation holds %s; remove it only if that process is gone", l.path) + return nil, fmt.Errorf("another boatstack-reviewer invocation holds %s; remove it only if that process is gone", path) } return nil, err } - return directoryLock{path: l.path}, nil + return directoryLock{path: path}, nil } type directoryLock struct{ path string } diff --git a/boatstack/conformance/behavior/kernel_backend_test.go b/boatstack/conformance/behavior/kernel_backend_test.go index 3929264..e0a47d3 100644 --- a/boatstack/conformance/behavior/kernel_backend_test.go +++ b/boatstack/conformance/behavior/kernel_backend_test.go @@ -293,16 +293,16 @@ type kernelInstrumentedStore struct { events *eventLog } -func (s *kernelInstrumentedStore) BeginEffect(ctx context.Context, revision uint64, target kernel.ControlState) error { - if err := s.Store.BeginEffect(ctx, revision, target); err != nil { +func (s *kernelInstrumentedStore) BeginEffect(ctx context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { + if err := s.Store.BeginEffect(ctx, instanceID, revision, attempt); err != nil { return err } s.events.append(behavior.EventAttemptDurable) return nil } -func (s *kernelInstrumentedStore) CommitTransition(ctx context.Context, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { - if err := s.Store.CommitTransition(ctx, revision, target, receipt); err != nil { +func (s *kernelInstrumentedStore) CommitTransition(ctx context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { + if err := s.Store.CommitTransition(ctx, instanceID, revision, target, receipt); err != nil { return err } s.events.append(behavior.EventCommitted) @@ -429,16 +429,16 @@ type sameBaseStore struct { barrier *twoPartyBarrier } -func (s *sameBaseStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { - state, err := s.Store.Load(ctx, instanceID) +func (s *sameBaseStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) barrierErr := s.barrier.wait() if err != nil { - return kernel.ControlState{}, err + return kernel.InstanceRecord{}, err } if barrierErr != nil { - return kernel.ControlState{}, barrierErr + return kernel.InstanceRecord{}, barrierErr } - return state, nil + return record, nil } type sameBaseDomain struct { diff --git a/boatstack/kernel/conformance/conformance.go b/boatstack/kernel/conformance/conformance.go index 2a39f65..379bf85 100644 --- a/boatstack/kernel/conformance/conformance.go +++ b/boatstack/kernel/conformance/conformance.go @@ -636,16 +636,16 @@ type sameBaseLoadStore struct { barrier *twoPartyBarrier } -func (s *sameBaseLoadStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { - state, err := s.Store.Load(ctx, instanceID) +func (s *sameBaseLoadStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) barrierErr := s.barrier.wait() if err != nil { - return kernel.ControlState{}, err + return kernel.InstanceRecord{}, err } if barrierErr != nil { - return kernel.ControlState{}, barrierErr + return kernel.InstanceRecord{}, barrierErr } - return state, nil + return record, nil } type sameBaseObservationDomain struct { diff --git a/boatstack/kernel/conformance/conformance_test.go b/boatstack/kernel/conformance/conformance_test.go index 7962d15..b6d4bab 100644 --- a/boatstack/kernel/conformance/conformance_test.go +++ b/boatstack/kernel/conformance/conformance_test.go @@ -131,7 +131,7 @@ func TestCommittedOutcomeRejectsPriorReceiptRewrite(t *testing.T) { if err != nil { t.Fatal(err) } - receipts := fixture.Store.(*MemoryStateStore).receipts + receipts := fixture.Store.(*MemoryStateStore).instanceReceipts() receipts.mu.Lock() receipts.values[0] = returned receipts.mu.Unlock() @@ -220,7 +220,7 @@ func TestCommittedOutcomeRejectsFalsePriorObservation(t *testing.T) { t.Fatal(err) } returned.ID = "rcp-" + identity - receipts := fixture.Store.(*MemoryStateStore).receipts + receipts := fixture.Store.(*MemoryStateStore).instanceReceipts() receipts.mu.Lock() receipts.values[len(receipts.values)-1] = returned receipts.mu.Unlock() @@ -261,7 +261,7 @@ func TestCommittedOutcomeRejectsObjectiveLineageSubstitution(t *testing.T) { t.Fatal(err) } returned.ID = "rcp-" + digest - receipts := fixture.Store.(*MemoryStateStore).receipts + receipts := fixture.Store.(*MemoryStateStore).instanceReceipts() receipts.mu.Lock() receipts.values[len(receipts.values)-1] = returned receipts.mu.Unlock() @@ -373,9 +373,7 @@ func integerPriorityCycleFixture() KernelConformance { if err != nil { panic(err) } - state, _ := fixture.Store.(*MemoryStateStore).snapshot() - state.Program = fixture.Program.Identity() - fixture.Store.(*MemoryStateStore).state = state + fixture.Store.(*MemoryStateStore).retargetProgram(fixture.Program.Identity()) return fixture } @@ -433,7 +431,7 @@ type effectfulLoadStore struct { mutate func() } -func (s effectfulLoadStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { +func (s effectfulLoadStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { s.mutate() return s.Store.Load(ctx, instanceID) } @@ -459,12 +457,8 @@ type noCASStore struct{ *MemoryStateStore } type blindCommitStore struct{ *MemoryStateStore } -func (s *blindCommitStore) CommitTransition(_ context.Context, _ uint64, target kernel.ControlState, receipt kernel.Receipt) error { - s.mu.Lock() - defer s.mu.Unlock() - s.state = cloneState(target) - s.commitCount++ - s.receipts.append(receipt) +func (s *blindCommitStore) CommitTransition(_ context.Context, instanceID string, _ uint64, target kernel.ControlState, receipt kernel.Receipt) error { + s.forceCommit(instanceID, target, receipt) return nil } @@ -484,42 +478,34 @@ type failOneLoadStore struct { failed bool } -func (s *failOneLoadStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { +func (s *failOneLoadStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { s.mu.Lock() if !s.failed { s.failed = true s.mu.Unlock() - return kernel.ControlState{}, fmt.Errorf("simulated asymmetric Load failure") + return kernel.InstanceRecord{}, fmt.Errorf("simulated asymmetric Load failure") } s.mu.Unlock() return s.Store.Load(ctx, instanceID) } -func (s *noCASStore) BeginEffect(_ context.Context, _ uint64, target kernel.ControlState) error { - s.mu.Lock() - defer s.mu.Unlock() - s.state = cloneState(target) +func (s *noCASStore) BeginEffect(_ context.Context, instanceID string, _ uint64, attempt kernel.ControlState) error { + s.forceState(instanceID, attempt) return nil } -func (s *noCASStore) CommitTransition(_ context.Context, _ uint64, target kernel.ControlState, receipt kernel.Receipt) error { - s.mu.Lock() - defer s.mu.Unlock() - s.state = cloneState(target) - s.commitCount++ - s.receipts.append(receipt) +func (s *noCASStore) CommitTransition(_ context.Context, instanceID string, _ uint64, target kernel.ControlState, receipt kernel.Receipt) error { + s.forceCommit(instanceID, target, receipt) return nil } -func (s tornCommitStore) CommitTransition(_ context.Context, _ uint64, target kernel.ControlState, _ kernel.Receipt) error { - s.mu.Lock() - defer s.mu.Unlock() - s.state.Mode = target.Mode +func (s tornCommitStore) CommitTransition(_ context.Context, instanceID string, _ uint64, target kernel.ControlState, _ kernel.Receipt) error { + s.forceMode(instanceID, target.Mode) return fmt.Errorf("simulated torn commit") } -func (s substitutingReceiptStore) CommitTransition(ctx context.Context, revision uint64, target kernel.ControlState, _ kernel.Receipt) error { - return s.Store.CommitTransition(ctx, revision, target, s.Receipt) +func (s substitutingReceiptStore) CommitTransition(ctx context.Context, instanceID string, revision uint64, target kernel.ControlState, _ kernel.Receipt) error { + return s.Store.CommitTransition(ctx, instanceID, revision, target, s.Receipt) } type clobberAfterCommitStore struct { @@ -535,11 +521,15 @@ func newClobberAfterCommitStore(base *MemoryStateStore) *clobberAfterCommitStore return &clobberAfterCommitStore{base: base, ready: make(chan struct{}), committed: make(chan struct{})} } -func (s *clobberAfterCommitStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { +func (s *clobberAfterCommitStore) Create(ctx context.Context, instanceID string, initial kernel.ControlState) error { + return s.base.Create(ctx, instanceID, initial) +} + +func (s *clobberAfterCommitStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { return s.base.Load(ctx, instanceID) } -func (s *clobberAfterCommitStore) BeginEffect(ctx context.Context, revision uint64, target kernel.ControlState) error { +func (s *clobberAfterCommitStore) BeginEffect(ctx context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { s.mu.Lock() s.arrivals++ arrival := s.arrivals @@ -549,17 +539,15 @@ func (s *clobberAfterCommitStore) BeginEffect(ctx context.Context, revision uint s.mu.Unlock() <-s.ready if arrival == 1 { - return s.base.BeginEffect(ctx, revision, target) + return s.base.BeginEffect(ctx, instanceID, revision, attempt) } <-s.committed - s.base.mu.Lock() - s.base.state = cloneState(target) - s.base.mu.Unlock() + s.base.forceState(instanceID, attempt) return fmt.Errorf("stale revision after clobber") } -func (s *clobberAfterCommitStore) CommitTransition(ctx context.Context, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { - err := s.base.CommitTransition(ctx, revision, target, receipt) +func (s *clobberAfterCommitStore) CommitTransition(ctx context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { + err := s.base.CommitTransition(ctx, instanceID, revision, target, receipt) if err == nil { s.closeOnce.Do(func() { close(s.committed) }) } diff --git a/boatstack/kernel/conformance/instance_store.go b/boatstack/kernel/conformance/instance_store.go new file mode 100644 index 0000000..146743c --- /dev/null +++ b/boatstack/kernel/conformance/instance_store.go @@ -0,0 +1,1005 @@ +package conformance + +import ( + "context" + "fmt" + "reflect" + "sync" + "testing" + "time" + + "github.com/operatorstack/boatstack/boatstack/kernel" +) + +const ( + instanceBindTransition = "instance.bind" + instanceStepTransition = "instance.step" + instanceRecoverTransition = "instance.recover" +) + +// InstanceStoreHarness supplies one isolated persistence substrate to the +// instance-store laws. The suite owns its own reference program, domain, and +// operator; a fixture only provides the store, a per-instance locker, and a +// way to reopen a fresh handle over the same durable substrate. +type InstanceStoreHarness struct { + Store kernel.Store + Locker func() kernel.Locker + Reopen func(testing.TB) kernel.Store +} + +// InstanceStoreConformance proves a Store implementation supports durable, +// isolated, and idempotently resumable control instances: explicit atomic +// provisioning, atomic state-plus-history records, per-instance revision CAS +// and locking, append-only history, exact committed-result reconciliation, +// fail-closed corrupted-history handling, and restart reconstruction. +type InstanceStoreConformance struct { + New func(testing.TB) InstanceStoreHarness +} + +// Run executes every instance-store law against a fresh harness each. +func (suite InstanceStoreConformance) Run(t *testing.T) { + laws := []struct { + name string + law func(testing.TB, *instanceStoreFixture) error + }{ + {"create_load_restart", instanceProvisioningError}, + {"duplicate_and_concurrent_creation", instanceCreationRaceError}, + {"two_instances_isolated", instanceIsolationError}, + {"per_instance_cas", instanceCASError}, + {"per_instance_locking", instanceLockingError}, + {"atomic_commit_append_only_history", instanceAtomicHistoryError}, + {"committed_retry_returns_original_receipt", instanceReconciliationError}, + {"failed_attempt_retry_remains_recovery_required", instanceFailedAttemptError}, + {"cross_instance_prescription_replay_rejected", instanceCrossReplayError}, + {"concurrent_retries_return_one_durable_result", instanceConcurrentRetryError}, + {"corrupted_history_fails_closed", instanceHistoryFailClosedError}, + {"restart_reconstruction", instanceRestartError}, + } + for _, law := range laws { + law := law + t.Run(law.name, func(t *testing.T) { + fixture := newInstanceStoreFixture(t, suite.New(t)) + if err := law.law(t, fixture); err != nil { + t.Fatalf("control-law instance-store-%s: %v", law.name, err) + } + }) + } +} + +// KeyedMemoryLocker serializes each control instance independently so +// independent instances can progress concurrently. +type KeyedMemoryLocker struct { + mu sync.Mutex + locks map[string]*sync.Mutex +} + +// NewKeyedMemoryLocker returns an empty per-instance locker. +func NewKeyedMemoryLocker() *KeyedMemoryLocker { + return &KeyedMemoryLocker{locks: map[string]*sync.Mutex{}} +} + +func (l *KeyedMemoryLocker) Acquire(_ context.Context, instanceID string) (kernel.Lock, error) { + l.mu.Lock() + lock, ok := l.locks[instanceID] + if !ok { + lock = &sync.Mutex{} + l.locks[instanceID] = lock + } + l.mu.Unlock() + lock.Lock() + return memoryLock{mu: lock}, nil +} + +// instanceStoreDomain is the suite-owned counting domain: one integer value +// per control instance, with observation and verification counters so a law +// can prove a reconciled retry re-observed or re-verified nothing. +type instanceStoreDomain struct { + mu sync.Mutex + values map[string]int + observations map[string]int + verifications map[string]int +} + +func newInstanceStoreDomain() *instanceStoreDomain { + return &instanceStoreDomain{values: map[string]int{}, observations: map[string]int{}, verifications: map[string]int{}} +} + +func (d *instanceStoreDomain) Observe(_ context.Context, instanceID string) (kernel.Observation, error) { + d.mu.Lock() + d.observations[instanceID]++ + value := d.values[instanceID] + d.mu.Unlock() + return kernel.NewObservation(struct { + Instance string `json:"instance"` + Value int `json:"value"` + }{instanceID, value}) +} + +func (d *instanceStoreDomain) Admissible(_ context.Context, evaluation kernel.Evaluation) (bool, string, error) { + switch evaluation.Transition.Operation { + case instanceBindTransition: + return evaluation.State.ObjectiveBinding == nil && evaluation.Objective != nil, "initial binding requires an unbound instance and one exact objective", nil + case instanceStepTransition: + return true, "a step is always admissible", nil + case instanceRecoverTransition: + return evaluation.State.Recovery != nil, "an unresolved attempt requires recovery", nil + default: + return false, "unknown instance-store operation", nil + } +} + +func (d *instanceStoreDomain) Verify(_ context.Context, evaluation kernel.Evaluation, effect kernel.Effect, target kernel.Observation) error { + d.mu.Lock() + d.verifications[evaluation.State.InstanceID]++ + d.mu.Unlock() + if len(effect.Facts) != 1 { + return fmt.Errorf("instance-store effects carry exactly one fact") + } + return nil +} + +func (d *instanceStoreDomain) snapshot(instanceID string) (value, observations, verifications int) { + d.mu.Lock() + defer d.mu.Unlock() + return d.values[instanceID], d.observations[instanceID], d.verifications[instanceID] +} + +// instanceStoreOperator executes the suite operations and counts executions +// per instance, so a law can prove a retry executed no additional effect. +type instanceStoreOperator struct { + domain *instanceStoreDomain + mu sync.Mutex + executions map[string]int + interruptNext bool +} + +func newInstanceStoreOperator(domain *instanceStoreDomain) *instanceStoreOperator { + return &instanceStoreOperator{domain: domain, executions: map[string]int{}} +} + +func (o *instanceStoreOperator) Execute(_ context.Context, operation kernel.Operation) (kernel.Effect, error) { + o.mu.Lock() + o.executions[operation.InstanceID]++ + interrupt := o.interruptNext + o.interruptNext = false + o.mu.Unlock() + switch operation.Transition.Operation { + case instanceBindTransition: + if operation.Objective == nil { + return kernel.Effect{}, fmt.Errorf("initial binding lacks an objective") + } + return kernel.Effect{Facts: []kernel.EffectFact{{ + Facet: "supervisor.objective", Operation: instanceBindTransition, Fingerprint: operation.Objective.Fingerprint, + }}}, nil + case instanceStepTransition: + o.domain.mu.Lock() + o.domain.values[operation.InstanceID]++ + value := o.domain.values[operation.InstanceID] + o.domain.mu.Unlock() + if interrupt { + return kernel.Effect{}, fmt.Errorf("simulated interrupted operator") + } + return kernel.Effect{Facts: []kernel.EffectFact{{ + Facet: "instance.value", Operation: instanceStepTransition, Fingerprint: fmt.Sprintf("value-%d", value), + }}}, nil + case instanceRecoverTransition: + return kernel.Effect{Facts: []kernel.EffectFact{{ + Facet: "instance.value", Operation: instanceRecoverTransition, Fingerprint: "recovery-complete", + }}}, nil + default: + return kernel.Effect{}, fmt.Errorf("unknown instance-store operation") + } +} + +func (o *instanceStoreOperator) interruptNextStep() { + o.mu.Lock() + defer o.mu.Unlock() + o.interruptNext = true +} + +func (o *instanceStoreOperator) executionCount(instanceID string) int { + o.mu.Lock() + defer o.mu.Unlock() + return o.executions[instanceID] +} + +type instanceStoreCapabilities struct{} + +func (instanceStoreCapabilities) RequiredCapabilities(transition kernel.Transition) ([]kernel.Capability, error) { + switch transition.Operation { + case instanceBindTransition: + return []kernel.Capability{"objective.bind", "instance.apply"}, nil + case instanceStepTransition: + return []kernel.Capability{"instance.apply"}, nil + case instanceRecoverTransition: + return []kernel.Capability{"instance.recover"}, nil + default: + return nil, fmt.Errorf("unclassified instance-store operation %q", transition.Operation) + } +} + +func compileInstanceStoreProgram() (kernel.Program, error) { + return kernel.CompileProgram("instance-store", "1.0.0", "kernel-v1", "supervising", []string{"settled"}, []kernel.Transition{ + { + ID: instanceBindTransition, SourceModes: []string{"supervising"}, TargetMode: "supervising", + ObjectiveScope: kernel.ObjectiveNone, ObjectiveMutation: kernel.BindInitialObjective, + RequiredCapabilities: []kernel.Capability{"objective.bind"}, + OwnedFacets: []string{"supervisor.objective"}, + Operation: instanceBindTransition, SelectionRank: 1, Selection: kernel.SelectionExplicitOnly, Priority: 1, + }, + { + ID: instanceStepTransition, SourceModes: []string{"supervising"}, TargetMode: "supervising", + ObjectiveScope: kernel.ObjectiveOptionalPreserve, ObjectiveMutation: kernel.PreserveObjective, + RequiredCapabilities: []kernel.Capability{"instance.apply"}, + OwnedFacets: []string{"instance.value"}, + Operation: instanceStepTransition, SelectionRank: 2, Selection: kernel.SelectionExplicitOnly, Priority: 2, + }, + { + ID: instanceRecoverTransition, SourceModes: []string{"supervising"}, TargetMode: "supervising", + ObjectiveScope: kernel.ObjectiveOptionalPreserve, ObjectiveMutation: kernel.PreserveObjective, + RequiredCapabilities: []kernel.Capability{"instance.recover"}, + OwnedFacets: []string{"instance.value"}, + Operation: instanceRecoverTransition, SelectionRank: 1, Selection: kernel.SelectionImplicit, Priority: 10, + Recovers: []string{instanceBindTransition, instanceStepTransition, instanceRecoverTransition}, + }, + }) +} + +// instanceStoreFixture binds the suite-owned control program, domain, and +// operator to one harness substrate. +type instanceStoreFixture struct { + harness InstanceStoreHarness + program kernel.Program + domain *instanceStoreDomain + operator *instanceStoreOperator + classifier instanceStoreCapabilities + authority kernel.Authority + clock settlementClock +} + +func newInstanceStoreFixture(t testing.TB, harness InstanceStoreHarness) *instanceStoreFixture { + t.Helper() + if harness.Store == nil || harness.Locker == nil || harness.Reopen == nil { + t.Fatal("instance-store harness requires a store, a per-instance locker factory, and a reopen function") + } + program, err := compileInstanceStoreProgram() + if err != nil { + t.Fatal(err) + } + now := time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC) + domain := newInstanceStoreDomain() + return &instanceStoreFixture{ + harness: harness, + program: program, + domain: domain, + operator: newInstanceStoreOperator(domain), + authority: kernel.Authority{Receipts: []kernel.AuthorityReceipt{{ + ID: "instance-store-authority", Subject: "fixture", Fingerprint: "instance-store-authority", + Capabilities: []kernel.Capability{"instance.apply", "instance.recover", "objective.bind"}, + IssuedAt: now.Add(-time.Minute), ExpiresAt: now.Add(time.Hour), + }}}, + clock: settlementClock{now: now}, + } +} + +func (f *instanceStoreFixture) runtime(store kernel.Store, locker kernel.Locker) (kernel.Runtime, error) { + return kernel.NewRuntime(f.program, f.domain, f.operator, f.classifier, store, locker, f.clock) +} + +func (f *instanceStoreFixture) objective(instanceID string) (kernel.Objective, error) { + return kernel.NewObjective(instanceID+"-objective", 1, struct { + Instance string `json:"instance"` + }{instanceID}) +} + +func (f *instanceStoreFixture) resolve(runtime kernel.Runtime, instanceID, transition string, objective *kernel.Objective) (kernel.ResolveRequest, kernel.Prescription, error) { + request := kernel.ResolveRequest{InstanceID: instanceID, Objective: objective, Authority: f.authority, Requested: transition} + resolution, err := runtime.Resolve(context.Background(), request) + if err != nil { + return kernel.ResolveRequest{}, kernel.Prescription{}, err + } + if resolution.Decision.Kind != kernel.Prescribed || resolution.Prescription == nil { + return kernel.ResolveRequest{}, kernel.Prescription{}, fmt.Errorf("transition %q was not prescribed: %s", transition, resolution.Decision.Reason) + } + return request, *resolution.Prescription, nil +} + +func (f *instanceStoreFixture) apply(runtime kernel.Runtime, instanceID, transition string, objective *kernel.Objective) (kernel.Receipt, error) { + request, prescription, err := f.resolve(runtime, instanceID, transition, objective) + if err != nil { + return kernel.Receipt{}, err + } + return runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) +} + +// bindAndStep provisions and advances one instance through an initial +// binding plus one step, returning both committed receipts. +func (f *instanceStoreFixture) bindAndStep(runtime kernel.Runtime, instanceID string) (bind, step kernel.Receipt, err error) { + if _, err = runtime.Provision(context.Background(), instanceID); err != nil { + return kernel.Receipt{}, kernel.Receipt{}, err + } + objective, err := f.objective(instanceID) + if err != nil { + return kernel.Receipt{}, kernel.Receipt{}, err + } + bind, err = f.apply(runtime, instanceID, instanceBindTransition, &objective) + if err != nil { + return kernel.Receipt{}, kernel.Receipt{}, err + } + step, err = f.apply(runtime, instanceID, instanceStepTransition, nil) + return bind, step, err +} + +// instanceProvisioningError proves explicit atomic creation: a provisioned +// instance loads with its exact initial state and empty history, survives a +// reopen, duplicates are typed rejections, and a missing instance stays a +// typed not-found result. +func instanceProvisioningError(t testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + if _, err := fixture.harness.Store.Load(context.Background(), "instance-alpha"); !kernel.IsInstanceNotFound(err) { + return fmt.Errorf("missing instance load = %v, want typed not-found", err) + } + if _, err := runtime.Resolve(context.Background(), kernel.ResolveRequest{InstanceID: "instance-alpha", Authority: fixture.authority}); !kernel.IsInstanceNotFound(err) { + return fmt.Errorf("missing instance resolve = %v, want typed not-found", err) + } + initial, err := runtime.Provision(context.Background(), "instance-alpha") + if err != nil { + return fmt.Errorf("provisioning: %w", err) + } + if initial.InstanceID != "instance-alpha" || initial.Mode != fixture.program.InitialMode || initial.Revision != 1 || initial.ObjectiveBinding != nil || initial.Recovery != nil { + return fmt.Errorf("initial state %#v does not carry exact instance identity, initial mode, revision 1, and no manufactured objective state", initial) + } + if initial.Program != fixture.program.Identity() { + return fmt.Errorf("initial state program %#v is not the active program identity", initial.Program) + } + record, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if !reflect.DeepEqual(record.State, initial) || len(record.Receipts) != 0 { + return fmt.Errorf("loaded record %#v differs from the provisioned initial state with empty history", record) + } + if _, err := runtime.Provision(context.Background(), "instance-alpha"); !kernel.IsInstanceExists(err) { + return fmt.Errorf("duplicate provisioning = %v, want typed instance-exists", err) + } + reopened, err := fixture.harness.Reopen(t).Load(context.Background(), "instance-alpha") + if err != nil { + return fmt.Errorf("reopened load: %w", err) + } + if !reflect.DeepEqual(reopened, record) { + return fmt.Errorf("reopened record %#v differs from the created record %#v", reopened, record) + } + return nil +} + +// instanceCreationRaceError proves concurrent creation of one identity +// yields exactly one initial history and one typed loser. +func instanceCreationRaceError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + start := make(chan struct{}) + results := make(chan error, 2) + for range 2 { + go func() { + <-start + _, provisionErr := runtime.Provision(context.Background(), "instance-race") + results <- provisionErr + }() + } + close(start) + var winners, losers int + for range 2 { + switch err := <-results; { + case err == nil: + winners++ + case kernel.IsInstanceExists(err): + losers++ + default: + return fmt.Errorf("concurrent creation = %v, want nil or typed instance-exists", err) + } + } + if winners != 1 || losers != 1 { + return fmt.Errorf("concurrent creation produced %d winners and %d typed losers, want exactly one of each", winners, losers) + } + record, err := fixture.harness.Store.Load(context.Background(), "instance-race") + if err != nil { + return err + } + if record.State.Revision != 1 || len(record.Receipts) != 0 { + return fmt.Errorf("concurrent creation produced record %#v, want one initial history", record) + } + return nil +} + +// instanceIsolationError proves two independent objectives occupy two +// durable instances in one store without any cross-instance effect: state, +// revisions, receipts, and receipt instance identity stay local. +func instanceIsolationError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + if _, _, err := fixture.bindAndStep(runtime, "instance-alpha"); err != nil { + return err + } + if _, err := runtime.Provision(context.Background(), "instance-beta"); err != nil { + return err + } + beta, err := fixture.harness.Store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + if beta.State.Revision != 1 || beta.State.ObjectiveBinding != nil || len(beta.Receipts) != 0 { + return fmt.Errorf("progress on instance-alpha leaked into instance-beta: %#v", beta) + } + alpha, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if len(alpha.Receipts) != 2 { + return fmt.Errorf("instance-alpha history has %d receipts, want 2", len(alpha.Receipts)) + } + for _, receipt := range alpha.Receipts { + if receipt.InstanceID != "instance-alpha" { + return fmt.Errorf("committed receipt %q carries instance %q, not its record's instance", receipt.ID, receipt.InstanceID) + } + } + betaObjective, err := fixture.objective("instance-beta") + if err != nil { + return err + } + if _, err := fixture.apply(runtime, "instance-beta", instanceBindTransition, &betaObjective); err != nil { + return err + } + after, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if !reflect.DeepEqual(after, alpha) { + return fmt.Errorf("progress on instance-beta mutated instance-alpha: before=%#v after=%#v", alpha, after) + } + alphaBinding, betaRecord := after.State.ObjectiveBinding, kernel.InstanceRecord{} + if betaRecord, err = fixture.harness.Store.Load(context.Background(), "instance-beta"); err != nil { + return err + } + if alphaBinding == nil || betaRecord.State.ObjectiveBinding == nil || *alphaBinding == *betaRecord.State.ObjectiveBinding { + return fmt.Errorf("the two instances do not hold two independent objectives: alpha=%#v beta=%#v", alphaBinding, betaRecord.State.ObjectiveBinding) + } + return nil +} + +// mintInstanceReceipt builds a content-addressed receipt the way the kernel +// does, so store-level laws can commit valid history without a runtime. +func mintInstanceReceipt(fixture *instanceStoreFixture, instanceID, prescriptionID string, priorRevision uint64) (kernel.Receipt, error) { + receipt := kernel.Receipt{ + SchemaVersion: kernel.ReceiptSchemaVersion, InstanceID: instanceID, + PrescriptionID: prescriptionID, Program: fixture.program.Identity(), TransitionID: instanceStepTransition, + ObjectiveMutation: kernel.PreserveObjective, + PriorStateRevision: priorRevision, AttemptStateRevision: priorRevision + 1, ResultStateRevision: priorRevision + 2, + AuthorityFingerprint: "instance-store-authority", + Capabilities: []kernel.Capability{"instance.apply"}, + Effects: []kernel.EffectFact{{Facet: "instance.value", Operation: instanceStepTransition, Fingerprint: "value-1"}}, + PriorObservation: fmt.Sprintf("%064d", 1), ResultObservation: fmt.Sprintf("%064d", 2), + Verification: "satisfied", CommittedAt: fixture.clock.Now().UTC(), + } + identity := receipt + identity.ID = "" + digest, err := kernel.Fingerprint(identity) + if err != nil { + return kernel.Receipt{}, err + } + receipt.ID = "rcp-" + digest + return receipt, nil +} + +// instanceCASError proves the revision compare-and-swap is local to one +// instance: stale attempts and commits are rejected, a failed final commit +// retains the unresolved attempt without a receipt, and another instance's +// revision is never consulted. +func instanceCASError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + for _, instanceID := range []string{"instance-alpha", "instance-beta"} { + if _, err := runtime.Provision(context.Background(), instanceID); err != nil { + return err + } + } + // Advance alpha so the two instances hold different revisions. + if _, err := fixture.apply(runtime, "instance-alpha", instanceStepTransition, nil); err != nil { + return err + } + store := fixture.harness.Store + alpha, err := store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if alpha.State.Revision != 3 { + return fmt.Errorf("instance-alpha revision is %d after one committed step, want 3", alpha.State.Revision) + } + staleAttempt := alpha.State + staleAttempt.Revision = 2 + if err := store.BeginEffect(context.Background(), "instance-alpha", 1, staleAttempt); err == nil { + return fmt.Errorf("stale effect attempt against instance-alpha was accepted") + } + // Beta's CAS must still expect beta's own revision 1, not alpha's 3. + beta, err := store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + attempt := beta.State + attempt.Revision = 2 + attempt.Recovery = &kernel.RecoveryState{PrescriptionID: "prx-cas-law", TransitionID: instanceStepTransition, Reason: "effect attempt began; outcome is unresolved"} + if err := store.BeginEffect(context.Background(), "instance-beta", 3, attempt); err == nil { + return fmt.Errorf("instance-beta accepted a compare-and-swap against another instance's revision") + } + if err := store.BeginEffect(context.Background(), "instance-beta", 1, attempt); err != nil { + return fmt.Errorf("instance-beta rejected its own current revision: %w", err) + } + target := attempt + target.Revision = 3 + target.Recovery = nil + receipt, err := mintInstanceReceipt(fixture, "instance-beta", "prx-cas-law", 1) + if err != nil { + return err + } + if err := store.CommitTransition(context.Background(), "instance-beta", 1, target, receipt); err == nil { + return fmt.Errorf("stale final commit against instance-beta was accepted") + } + after, err := store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + if !reflect.DeepEqual(after.State, attempt) || len(after.Receipts) != 0 { + return fmt.Errorf("failed final commit did not retain the unresolved attempt without a receipt: %#v", after) + } + if err := store.CommitTransition(context.Background(), "instance-beta", 2, target, receipt); err != nil { + return fmt.Errorf("current-revision commit against instance-beta failed: %w", err) + } + committed, err := store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + if committed.State.Revision != 3 || len(committed.Receipts) != 1 || committed.Receipts[0].ID != receipt.ID { + return fmt.Errorf("commit did not atomically persist target state plus exactly one receipt: %#v", committed) + } + return nil +} + +// instanceLockingError proves locking is per instance: holding one +// instance's lock does not block acquiring another's. +func instanceLockingError(_ testing.TB, fixture *instanceStoreFixture) error { + locker := fixture.harness.Locker() + alphaLock, err := locker.Acquire(context.Background(), "instance-alpha") + if err != nil { + return err + } + defer alphaLock.Unlock() + acquired := make(chan error, 1) + go func() { + betaLock, betaErr := locker.Acquire(context.Background(), "instance-beta") + if betaErr == nil { + betaLock.Unlock() + } + acquired <- betaErr + }() + select { + case err := <-acquired: + return err + case <-time.After(5 * time.Second): + return fmt.Errorf("holding instance-alpha's lock blocked instance-beta's lock; locking is not per instance") + } +} + +// instanceAtomicHistoryError proves history is append-only and every commit +// makes the target state and its receipt visible together. +func instanceAtomicHistoryError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + bind, step, err := fixture.bindAndStep(runtime, "instance-alpha") + if err != nil { + return err + } + record, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if len(record.Receipts) != 2 || record.Receipts[0].ID != bind.ID || record.Receipts[1].ID != step.ID { + return fmt.Errorf("history %#v is not the append-only ordered sequence of committed receipts", record.Receipts) + } + if record.State.Revision != step.ResultStateRevision { + return fmt.Errorf("state revision %d and final receipt result revision %d were not committed together", record.State.Revision, step.ResultStateRevision) + } + next, err := fixture.apply(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + after, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if len(after.Receipts) != 3 || after.Receipts[0].ID != bind.ID || after.Receipts[1].ID != step.ID || after.Receipts[2].ID != next.ID { + return fmt.Errorf("a later commit rewrote earlier history: %#v", after.Receipts) + } + return nil +} + +// instanceReconciliationError proves an exact committed retry returns the +// original durable receipt with no additional observation, execution, +// verification, state mutation, receipt, or revision — and that the lookup +// is by exact prescription, not by the latest receipt: retrying an earlier +// committed prescription returns that earlier receipt. +func instanceReconciliationError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + if _, err := runtime.Provision(context.Background(), "instance-alpha"); err != nil { + return err + } + objective, err := fixture.objective("instance-alpha") + if err != nil { + return err + } + bindRequest, bindPrescription, err := fixture.resolve(runtime, "instance-alpha", instanceBindTransition, &objective) + if err != nil { + return err + } + bindReceipt, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: bindRequest, Prescription: bindPrescription}) + if err != nil { + return err + } + stepRequest, stepPrescription, err := fixture.resolve(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + stepReceipt, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: stepRequest, Prescription: stepPrescription}) + if err != nil { + return err + } + before, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + value, observations, verifications := fixture.domain.snapshot("instance-alpha") + executions := fixture.operator.executionCount("instance-alpha") + retries := []struct { + name string + request kernel.ResolveRequest + presc kernel.Prescription + original kernel.Receipt + }{ + {"latest committed prescription", stepRequest, stepPrescription, stepReceipt}, + {"earlier committed prescription", bindRequest, bindPrescription, bindReceipt}, + } + for _, retry := range retries { + reconciled, applyErr := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: retry.request, Prescription: retry.presc}) + if applyErr != nil { + return fmt.Errorf("retry of the %s = %v, want its original durable receipt", retry.name, applyErr) + } + if !reflect.DeepEqual(reconciled, retry.original) { + return fmt.Errorf("retry of the %s returned a different receipt: original=%#v reconciled=%#v", retry.name, retry.original, reconciled) + } + } + afterValue, afterObservations, afterVerifications := fixture.domain.snapshot("instance-alpha") + if afterValue != value || afterObservations != observations || afterVerifications != verifications || fixture.operator.executionCount("instance-alpha") != executions { + return fmt.Errorf("reconciliation re-observed, re-executed, or re-verified: observations %d→%d executions %d→%d verifications %d→%d", observations, afterObservations, executions, fixture.operator.executionCount("instance-alpha"), verifications, afterVerifications) + } + after, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if !reflect.DeepEqual(after, before) { + return fmt.Errorf("reconciliation mutated durable state or history: before=%#v after=%#v", before, after) + } + return nil +} + +// instanceFailedAttemptError proves a failed or interrupted attempt without +// a committed receipt is never reported as successful: the retry stays +// recovery-required and only the declared recovery transition resolves it. +func instanceFailedAttemptError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + if _, err := runtime.Provision(context.Background(), "instance-alpha"); err != nil { + return err + } + request, prescription, err := fixture.resolve(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + fixture.operator.interruptNextStep() + if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}); !kernel.IsRecoveryRequired(err) { + return fmt.Errorf("interrupted attempt = %v, want recovery required", err) + } + record, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if record.State.Recovery == nil || record.State.Recovery.PrescriptionID != prescription.ID || len(record.Receipts) != 0 { + return fmt.Errorf("interrupted attempt did not persist an unresolved recovery obligation: %#v", record) + } + retried, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err == nil { + return fmt.Errorf("failed-attempt retry reported synthetic success: %#v", retried) + } + if !kernel.IsRecoveryRequired(err) { + return fmt.Errorf("failed-attempt retry = %v, want recovery required", err) + } + if _, err := fixture.apply(runtime, "instance-alpha", instanceRecoverTransition, nil); err != nil { + return fmt.Errorf("declared recovery transition: %w", err) + } + recovered, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if recovered.State.Recovery != nil || len(recovered.Receipts) != 1 || recovered.Receipts[0].TransitionID != instanceRecoverTransition { + return fmt.Errorf("recovery did not settle the obligation with exactly one recovery receipt: %#v", recovered) + } + return nil +} + +// instanceCrossReplayError proves a prescription minted for one instance is +// rejected against another before any effect, leaving the target untouched. +func instanceCrossReplayError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + for _, instanceID := range []string{"instance-alpha", "instance-beta"} { + if _, err := runtime.Provision(context.Background(), instanceID); err != nil { + return err + } + } + request, prescription, err := fixture.resolve(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + before, err := fixture.harness.Store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + executions := fixture.operator.executionCount("instance-beta") + request.InstanceID = "instance-beta" + if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}); !kernel.IsStale(err) { + return fmt.Errorf("cross-instance replay = %v, want stale rejection", err) + } + after, err := fixture.harness.Store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + if !reflect.DeepEqual(after, before) || fixture.operator.executionCount("instance-beta") != executions { + return fmt.Errorf("cross-instance replay mutated the target instance: before=%#v after=%#v", before, after) + } + return nil +} + +// instanceConcurrentRetryError proves concurrent retries of one committed +// prescription reconcile to the single durable result. A non-blocking locker +// may refuse one contender, but every retry that completes must return the +// original receipt, at least one must complete, and durable state must not +// move. +func instanceConcurrentRetryError(_ testing.TB, fixture *instanceStoreFixture) error { + locker := fixture.harness.Locker() + runtime, err := fixture.runtime(fixture.harness.Store, locker) + if err != nil { + return err + } + if _, err := runtime.Provision(context.Background(), "instance-alpha"); err != nil { + return err + } + request, prescription, err := fixture.resolve(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + original, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { + return err + } + before, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + type retryResult struct { + receipt kernel.Receipt + err error + } + results := make(chan retryResult, 2) + for range 2 { + go func() { + receipt, applyErr := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + results <- retryResult{receipt: receipt, err: applyErr} + }() + } + reconciliations := 0 + for range 2 { + result := <-results + if result.err != nil { + continue + } + reconciliations++ + if !reflect.DeepEqual(result.receipt, original) { + return fmt.Errorf("concurrent retry returned a different receipt: original=%#v got=%#v", original, result.receipt) + } + } + if reconciliations == 0 { + return fmt.Errorf("no concurrent retry reconciled to the original durable receipt") + } + after, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + if !reflect.DeepEqual(after, before) { + return fmt.Errorf("concurrent retries mutated durable state: before=%#v after=%#v", before, after) + } + return nil +} + +// corruptingLoadStore corrupts every loaded record, standing in for a store +// whose durable history was substituted, duplicated, reordered, misrouted, +// or truncated after restart. +type corruptingLoadStore struct { + kernel.Store + corrupt func(kernel.InstanceRecord) kernel.InstanceRecord +} + +func (s corruptingLoadStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) + if err != nil { + return kernel.InstanceRecord{}, err + } + return s.corrupt(record), nil +} + +// instanceHistoryFailClosedError proves the runtime fails closed on +// substituted, duplicated, reordered, misrouted, and truncated history: no +// corruption can produce a successful committed-retry reconciliation. +func instanceHistoryFailClosedError(_ testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + if _, _, err := fixture.bindAndStep(runtime, "instance-alpha"); err != nil { + return err + } + betaRuntime := runtime + if _, err := betaRuntime.Provision(context.Background(), "instance-beta"); err != nil { + return err + } + betaRequest, betaPrescription, err := fixture.resolve(runtime, "instance-beta", instanceStepTransition, nil) + if err != nil { + return err + } + misrouted, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: betaRequest, Prescription: betaPrescription}) + if err != nil { + return err + } + request, prescription, err := fixture.resolve(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + committed, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { + return err + } + corruptions := map[string]func(kernel.InstanceRecord) kernel.InstanceRecord{ + "substituted_receipt": func(record kernel.InstanceRecord) kernel.InstanceRecord { + record.Receipts = append([]kernel.Receipt(nil), record.Receipts...) + record.Receipts[len(record.Receipts)-1] = misrouted + return record + }, + "duplicated_receipt": func(record kernel.InstanceRecord) kernel.InstanceRecord { + record.Receipts = append(append([]kernel.Receipt(nil), record.Receipts...), record.Receipts[len(record.Receipts)-1]) + return record + }, + "reordered_receipts": func(record kernel.InstanceRecord) kernel.InstanceRecord { + record.Receipts = append([]kernel.Receipt(nil), record.Receipts...) + record.Receipts[0], record.Receipts[len(record.Receipts)-1] = record.Receipts[len(record.Receipts)-1], record.Receipts[0] + return record + }, + "misrouted_receipt": func(record kernel.InstanceRecord) kernel.InstanceRecord { + record.Receipts = append([]kernel.Receipt(nil), record.Receipts...) + record.Receipts = append(record.Receipts, misrouted) + return record + }, + "truncated_history": func(record kernel.InstanceRecord) kernel.InstanceRecord { + record.Receipts = nil + return record + }, + } + for name, corrupt := range corruptions { + corrupted, runtimeErr := fixture.runtime(corruptingLoadStore{Store: fixture.harness.Store, corrupt: corrupt}, fixture.harness.Locker()) + if runtimeErr != nil { + return runtimeErr + } + receipt, applyErr := corrupted.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if applyErr == nil { + return fmt.Errorf("%s history reconciled a committed retry successfully: %#v", name, receipt) + } + if reflect.DeepEqual(receipt, committed) { + return fmt.Errorf("%s history still returned the original receipt", name) + } + } + return nil +} + +// instanceRestartError proves a reopened substrate reconstructs exact state, +// recovery, and history for independent instances, and still reconciles a +// lost successful response without executing the accepted transition twice. +func instanceRestartError(t testing.TB, fixture *instanceStoreFixture) error { + runtime, err := fixture.runtime(fixture.harness.Store, fixture.harness.Locker()) + if err != nil { + return err + } + if _, _, err := fixture.bindAndStep(runtime, "instance-alpha"); err != nil { + return err + } + request, prescription, err := fixture.resolve(runtime, "instance-alpha", instanceStepTransition, nil) + if err != nil { + return err + } + lost, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { + return err + } + if _, err := runtime.Provision(context.Background(), "instance-beta"); err != nil { + return err + } + betaRequest, betaPrescription, err := fixture.resolve(runtime, "instance-beta", instanceStepTransition, nil) + if err != nil { + return err + } + fixture.operator.interruptNextStep() + if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: betaRequest, Prescription: betaPrescription}); !kernel.IsRecoveryRequired(err) { + return fmt.Errorf("interrupted attempt on instance-beta = %v, want recovery required", err) + } + alphaBefore, err := fixture.harness.Store.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + betaBefore, err := fixture.harness.Store.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + reopened := fixture.harness.Reopen(t) + alphaAfter, err := reopened.Load(context.Background(), "instance-alpha") + if err != nil { + return err + } + betaAfter, err := reopened.Load(context.Background(), "instance-beta") + if err != nil { + return err + } + if !reflect.DeepEqual(alphaAfter, alphaBefore) || !reflect.DeepEqual(betaAfter, betaBefore) { + return fmt.Errorf("restart did not reconstruct exact records: alpha before=%#v after=%#v beta before=%#v after=%#v", alphaBefore, alphaAfter, betaBefore, betaAfter) + } + if betaAfter.State.Recovery == nil || betaAfter.State.Recovery.PrescriptionID != betaPrescription.ID { + return fmt.Errorf("restart did not reconstruct instance-beta's unresolved recovery obligation: %#v", betaAfter.State) + } + restarted, err := fixture.runtime(reopened, fixture.harness.Locker()) + if err != nil { + return err + } + executions := fixture.operator.executionCount("instance-alpha") + reconciled, err := restarted.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { + return fmt.Errorf("lost-response reconciliation after restart = %v, want the original durable receipt", err) + } + if !reflect.DeepEqual(reconciled, lost) { + return fmt.Errorf("restart reconciliation returned a different receipt: original=%#v reconciled=%#v", lost, reconciled) + } + if fixture.operator.executionCount("instance-alpha") != executions { + return fmt.Errorf("restart reconciliation executed the accepted transition again") + } + return nil +} diff --git a/boatstack/kernel/conformance/instance_store_law_test.go b/boatstack/kernel/conformance/instance_store_law_test.go new file mode 100644 index 0000000..977c863 --- /dev/null +++ b/boatstack/kernel/conformance/instance_store_law_test.go @@ -0,0 +1,275 @@ +package conformance + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/operatorstack/boatstack/boatstack/kernel" +) + +// memoryInstanceHarness runs the instance-store laws over the integer +// fixture's keyed in-memory store. Memory is the durable substrate, so +// reopening returns the same store handle. +func memoryInstanceHarness(testing.TB) InstanceStoreHarness { + store := NewMemoryStateStore() + return InstanceStoreHarness{ + Store: store, + Locker: func() kernel.Locker { return NewKeyedMemoryLocker() }, + Reopen: func(testing.TB) kernel.Store { return store }, + } +} + +func TestInstanceStoreConformanceMemoryStore(t *testing.T) { + InstanceStoreConformance{New: memoryInstanceHarness}.Run(t) +} + +func TestInstanceStoreConformanceRegisterStore(t *testing.T) { + InstanceStoreConformance{New: func(testing.TB) InstanceStoreHarness { + store := ®isterStore{instances: map[string]*registerInstance{}} + return InstanceStoreHarness{ + Store: store, + Locker: func() kernel.Locker { return NewKeyedMemoryLocker() }, + Reopen: func(testing.TB) kernel.Store { return store }, + } + }}.Run(t) +} + +// singletonInstanceStore dishonestly keeps exactly one record and ignores +// the requested instance identity everywhere: creation of a second instance +// is silently absorbed and every load returns the one record. +type singletonInstanceStore struct { + mu sync.Mutex + created bool + state kernel.ControlState + receipts []kernel.Receipt +} + +func (s *singletonInstanceStore) Create(_ context.Context, _ string, initial kernel.ControlState) error { + s.mu.Lock() + defer s.mu.Unlock() + if !s.created { + s.state = cloneState(initial) + s.created = true + } + return nil +} + +func (s *singletonInstanceStore) Load(_ context.Context, instanceID string) (kernel.InstanceRecord, error) { + s.mu.Lock() + defer s.mu.Unlock() + if !s.created { + return kernel.InstanceRecord{}, kernel.InstanceNotFoundError{InstanceID: instanceID} + } + return kernel.InstanceRecord{State: cloneState(s.state), Receipts: append([]kernel.Receipt(nil), s.receipts...)}, nil +} + +func (s *singletonInstanceStore) BeginEffect(_ context.Context, _ string, revision uint64, attempt kernel.ControlState) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state.Revision != revision { + return fmt.Errorf("stale revision") + } + s.state = cloneState(attempt) + return nil +} + +func (s *singletonInstanceStore) CommitTransition(_ context.Context, _ string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state.Revision != revision { + return fmt.Errorf("stale revision") + } + s.state = cloneState(target) + s.receipts = append(s.receipts, receipt) + return nil +} + +// tornInstanceCommitStore dishonestly persists the target state without its +// receipt, tearing the atomic state-plus-receipt commit. +type tornInstanceCommitStore struct{ *MemoryStateStore } + +func (s tornInstanceCommitStore) CommitTransition(ctx context.Context, instanceID string, revision uint64, target kernel.ControlState, _ kernel.Receipt) error { + record, err := s.MemoryStateStore.Load(ctx, instanceID) + if err != nil { + return err + } + if record.State.Revision != revision { + return fmt.Errorf("stale revision") + } + s.forceState(instanceID, target) + return nil +} + +// blindGlobalCASStore dishonestly compares every attempt and commit against +// one store-global revision instead of the requested instance's revision. +type blindGlobalCASStore struct { + *MemoryStateStore + mu sync.Mutex + revision uint64 +} + +func (s *blindGlobalCASStore) Create(ctx context.Context, instanceID string, initial kernel.ControlState) error { + if err := s.MemoryStateStore.Create(ctx, instanceID, initial); err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + if s.revision == 0 { + s.revision = initial.Revision + } + return nil +} + +func (s *blindGlobalCASStore) BeginEffect(_ context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.revision != revision { + return fmt.Errorf("stale revision") + } + s.revision = attempt.Revision + s.forceState(instanceID, attempt) + return nil +} + +func (s *blindGlobalCASStore) CommitTransition(_ context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.revision != revision { + return fmt.Errorf("stale revision") + } + s.revision = target.Revision + s.forceCommit(instanceID, target, receipt) + return nil +} + +// latestReceiptStore dishonestly exposes only the newest committed receipt, +// standing in for a store that reconciles by latest receipt rather than by +// exact prescription. +type latestReceiptStore struct{ kernel.Store } + +func (s latestReceiptStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) + if err != nil { + return kernel.InstanceRecord{}, err + } + if len(record.Receipts) > 1 { + record.Receipts = append([]kernel.Receipt(nil), record.Receipts[len(record.Receipts)-1]) + } + return record, nil +} + +// remintReceipt recomputes a receipt's content-addressed identity after a +// dishonest mutation, so the fabrication survives content validation. +func remintReceipt(receipt kernel.Receipt) kernel.Receipt { + identity := receipt + identity.ID = "" + digest, err := kernel.Fingerprint(identity) + if err != nil { + panic(err) + } + receipt.ID = "rcp-" + digest + return receipt +} + +// replacingHistoryStore dishonestly replaces the committed history: every +// loaded receipt is re-minted under a different prescription identity, so no +// exact committed request can ever be found again. +type replacingHistoryStore struct{ kernel.Store } + +func (s replacingHistoryStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) + if err != nil { + return kernel.InstanceRecord{}, err + } + replaced := make([]kernel.Receipt, len(record.Receipts)) + for index, receipt := range record.Receipts { + receipt.PrescriptionID = fmt.Sprintf("prx-replaced-%d", index) + replaced[index] = remintReceipt(receipt) + } + record.Receipts = replaced + return record, nil +} + +// truncatingHistoryStore dishonestly drops the committed history while +// keeping the advanced control state. +type truncatingHistoryStore struct{ kernel.Store } + +func (s truncatingHistoryStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) + if err != nil { + return kernel.InstanceRecord{}, err + } + record.Receipts = nil + return record, nil +} + +// syntheticSuccessStore dishonestly converts an unresolved attempt into a +// fabricated success: it clears the recovery obligation, advances the state, +// and mints a coherent-looking receipt for the pending prescription. +type syntheticSuccessStore struct{ kernel.Store } + +func (s syntheticSuccessStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { + record, err := s.Store.Load(ctx, instanceID) + if err != nil { + return kernel.InstanceRecord{}, err + } + recovery := record.State.Recovery + if recovery == nil { + return record, nil + } + state := record.State + state.Recovery = nil + state.Revision++ + fabricated := remintReceipt(kernel.Receipt{ + SchemaVersion: kernel.ReceiptSchemaVersion, InstanceID: instanceID, + PrescriptionID: recovery.PrescriptionID, Program: state.Program, TransitionID: recovery.TransitionID, + ObjectiveMutation: kernel.PreserveObjective, + PriorStateRevision: state.Revision - 2, AttemptStateRevision: state.Revision - 1, ResultStateRevision: state.Revision, + AuthorityFingerprint: "instance-store-authority", + Capabilities: []kernel.Capability{"instance.apply"}, + Effects: []kernel.EffectFact{{Facet: "instance.value", Operation: recovery.TransitionID, Fingerprint: "value-synthetic"}}, + PriorObservation: fmt.Sprintf("%064d", 0), ResultObservation: fmt.Sprintf("%064d", 1), + Verification: "satisfied", CommittedAt: time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC), + }) + record.State = state + record.Receipts = append(append([]kernel.Receipt(nil), record.Receipts...), fabricated) + return record, nil +} + +// TestInstanceStoreLawsRejectDishonestStores proves the suite is sharp: each +// dishonest white-box store fails the exact law that guards the behavior it +// forges. +func TestInstanceStoreLawsRejectDishonestStores(t *testing.T) { + cases := []struct { + name string + store func() kernel.Store + law func(testing.TB, *instanceStoreFixture) error + }{ + {"singleton_store_ignoring_the_requested_instance", func() kernel.Store { return &singletonInstanceStore{} }, instanceIsolationError}, + {"torn_state_plus_receipt_commit", func() kernel.Store { return tornInstanceCommitStore{NewMemoryStateStore()} }, instanceAtomicHistoryError}, + {"blind_global_compare_and_swap", func() kernel.Store { return &blindGlobalCASStore{MemoryStateStore: NewMemoryStateStore()} }, instanceCASError}, + {"latest_receipt_rather_than_exact_prescription_lookup", func() kernel.Store { return latestReceiptStore{NewMemoryStateStore()} }, instanceReconciliationError}, + {"replaced_history", func() kernel.Store { return replacingHistoryStore{NewMemoryStateStore()} }, instanceReconciliationError}, + {"truncated_history", func() kernel.Store { return truncatingHistoryStore{NewMemoryStateStore()} }, instanceAtomicHistoryError}, + {"synthetic_success_after_an_unresolved_attempt", func() kernel.Store { return syntheticSuccessStore{NewMemoryStateStore()} }, instanceFailedAttemptError}, + } + for _, dishonest := range cases { + dishonest := dishonest + t.Run(dishonest.name, func(t *testing.T) { + store := dishonest.store() + fixture := newInstanceStoreFixture(t, InstanceStoreHarness{ + Store: store, + Locker: func() kernel.Locker { return NewKeyedMemoryLocker() }, + Reopen: func(testing.TB) kernel.Store { return store }, + }) + err := dishonest.law(t, fixture) + if err == nil { + t.Fatalf("the dishonest store %q passed its guarding law", dishonest.name) + } + t.Logf("rejected: %v", err) + }) + } +} diff --git a/boatstack/kernel/conformance/integer.go b/boatstack/kernel/conformance/integer.go index db85d11..ff954d0 100644 --- a/boatstack/kernel/conformance/integer.go +++ b/boatstack/kernel/conformance/integer.go @@ -158,7 +158,8 @@ func (IntegerCapabilities) RequiredCapabilities(transition kernel.Transition) ([ } } -// MemoryReceipts records committed receipts for the reference store. +// MemoryReceipts records committed receipts for one control instance in the +// reference store. type MemoryReceipts struct { mu sync.Mutex values []kernel.Receipt @@ -176,51 +177,111 @@ func (r *MemoryReceipts) snapshot() []kernel.Receipt { return append([]kernel.Receipt(nil), r.values...) } -// MemoryStateStore is a revision-CAS reference Store. +// memoryInstance is one durable instance record in the reference store. +type memoryInstance struct { + state kernel.ControlState + receipts *MemoryReceipts + commits int +} + +// MemoryStateStore is the keyed multi-instance, revision-CAS reference +// Store. Every operation routes by the explicit instance identity; state, +// receipts, revisions, and compare-and-swap are local to one instance. type MemoryStateStore struct { mu sync.Mutex - state kernel.ControlState - receipts *MemoryReceipts + instances map[string]*memoryInstance + current string commitFailures int - commitCount int } -func (s *MemoryStateStore) Load(context.Context, string) (kernel.ControlState, error) { +// NewMemoryStateStore returns an empty multi-instance reference store. +func NewMemoryStateStore() *MemoryStateStore { + return &MemoryStateStore{instances: map[string]*memoryInstance{}} +} + +func (s *MemoryStateStore) Create(_ context.Context, instanceID string, initial kernel.ControlState) error { + s.mu.Lock() + defer s.mu.Unlock() + if initial.InstanceID != instanceID { + return fmt.Errorf("initial control state belongs to %q, not %q", initial.InstanceID, instanceID) + } + if _, ok := s.instances[instanceID]; ok { + return kernel.InstanceExistsError{InstanceID: instanceID} + } + s.instances[instanceID] = &memoryInstance{state: cloneState(initial), receipts: &MemoryReceipts{}} + if s.current == "" { + s.current = instanceID + } + return nil +} + +func (s *MemoryStateStore) Load(_ context.Context, instanceID string) (kernel.InstanceRecord, error) { s.mu.Lock() defer s.mu.Unlock() - return cloneState(s.state), nil + instance, ok := s.instances[instanceID] + if !ok { + return kernel.InstanceRecord{}, kernel.InstanceNotFoundError{InstanceID: instanceID} + } + return kernel.InstanceRecord{State: cloneState(instance.state), Receipts: instance.receipts.snapshot()}, nil } -func (s *MemoryStateStore) BeginEffect(_ context.Context, revision uint64, target kernel.ControlState) error { +func (s *MemoryStateStore) BeginEffect(_ context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { s.mu.Lock() defer s.mu.Unlock() - if s.state.Revision != revision { + instance, ok := s.instances[instanceID] + if !ok { + return kernel.InstanceNotFoundError{InstanceID: instanceID} + } + if instance.state.Revision != revision { return fmt.Errorf("stale revision") } - s.state = cloneState(target) + instance.state = cloneState(attempt) return nil } -func (s *MemoryStateStore) CommitTransition(_ context.Context, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { +func (s *MemoryStateStore) CommitTransition(_ context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { s.mu.Lock() defer s.mu.Unlock() + instance, ok := s.instances[instanceID] + if !ok { + return kernel.InstanceNotFoundError{InstanceID: instanceID} + } if s.commitFailures > 0 { s.commitFailures-- return fmt.Errorf("simulated atomic transaction failure") } - if s.state.Revision != revision { + if instance.state.Revision != revision { return fmt.Errorf("stale revision") } - s.state = cloneState(target) - s.commitCount++ - s.receipts.append(receipt) + instance.state = cloneState(target) + instance.commits++ + instance.receipts.append(receipt) return nil } +// currentInstance returns the fixture's active instance record. Fixture +// hooks act on this record; the kernel-facing Store methods never use it. +func (s *MemoryStateStore) currentInstance() *memoryInstance { + instance, ok := s.instances[s.current] + if !ok { + panic(fmt.Sprintf("memory store has no current instance %q", s.current)) + } + return instance +} + func (s *MemoryStateStore) snapshot() (kernel.ControlState, int) { s.mu.Lock() defer s.mu.Unlock() - return cloneState(s.state), s.commitCount + instance := s.currentInstance() + return cloneState(instance.state), instance.commits +} + +// instanceReceipts exposes the active instance's receipt log so white-box +// counterexample tests can rewrite committed history. +func (s *MemoryStateStore) instanceReceipts() *MemoryReceipts { + s.mu.Lock() + defer s.mu.Unlock() + return s.currentInstance().receipts } func (s *MemoryStateStore) failNextCommit() { @@ -229,28 +290,34 @@ func (s *MemoryStateStore) failNextCommit() { s.commitFailures++ } +// retarget provisions a separate control instance whose state mirrors the +// active one and makes it the fixture's active instance, so laws can address +// a prescription minted for one instance against another. func (s *MemoryStateStore) retarget(instanceID string) { s.mu.Lock() defer s.mu.Unlock() - s.state.InstanceID = instanceID + state := cloneState(s.currentInstance().state) + state.InstanceID = instanceID + s.instances[instanceID] = &memoryInstance{state: state, receipts: &MemoryReceipts{}} + s.current = instanceID } func (s *MemoryStateStore) isolate() { s.mu.Lock() defer s.mu.Unlock() - s.state.Mode = "isolated" + s.currentInstance().state.Mode = "isolated" } func (s *MemoryStateStore) bumpRevision() { s.mu.Lock() defer s.mu.Unlock() - s.state.Revision++ + s.currentInstance().state.Revision++ } func (s *MemoryStateStore) retargetProgram(program kernel.ProgramIdentity) { s.mu.Lock() defer s.mu.Unlock() - s.state.Program = program + s.currentInstance().state.Program = program } func (s *MemoryStateStore) rebind(objective kernel.Objective) { @@ -260,7 +327,34 @@ func (s *MemoryStateStore) rebind(objective kernel.Objective) { if err != nil { panic(err) } - s.state.ObjectiveBinding = &binding + s.currentInstance().state.ObjectiveBinding = &binding +} + +// forceState overwrites the requested instance's state without any +// compare-and-swap, for dishonest white-box store variants. +func (s *MemoryStateStore) forceState(instanceID string, state kernel.ControlState) { + s.mu.Lock() + defer s.mu.Unlock() + s.instances[instanceID].state = cloneState(state) +} + +// forceCommit commits state plus receipt without any compare-and-swap, for +// dishonest white-box store variants. +func (s *MemoryStateStore) forceCommit(instanceID string, target kernel.ControlState, receipt kernel.Receipt) { + s.mu.Lock() + defer s.mu.Unlock() + instance := s.instances[instanceID] + instance.state = cloneState(target) + instance.commits++ + instance.receipts.append(receipt) +} + +// forceMode tears a commit by mutating only the mode of the requested +// instance, for dishonest white-box store variants. +func (s *MemoryStateStore) forceMode(instanceID, mode string) { + s.mu.Lock() + defer s.mu.Unlock() + s.instances[instanceID].state.Mode = mode } // MemoryLocker serializes one control instance. @@ -365,8 +459,10 @@ func newIntegerFixture(setup Setup) KernelConformance { now := time.Date(2026, 8, 12, 10, 0, 0, 0, time.UTC) authority := kernel.Authority{Receipts: []kernel.AuthorityReceipt{{ID: "human-counter", Subject: "fixture", Fingerprint: "fixture-authority", Capabilities: []kernel.Capability{"counter.audit", "counter.hold", "counter.increment", "counter.reset", "objective.bind"}, IssuedAt: now.Add(-time.Minute), ExpiresAt: now.Add(24 * time.Hour)}}} domain := &IntegerDomain{value: value, executions: map[string]int{}} - receipts := &MemoryReceipts{} - store := &MemoryStateStore{state: state, receipts: receipts} + store := NewMemoryStateStore() + if err := store.Create(context.Background(), state.InstanceID, state); err != nil { + panic(err) + } clock := &FixedClock{Time: now} fixture := KernelConformance{ Domain: domain, @@ -411,7 +507,7 @@ func newIntegerFixture(setup Setup) KernelConformance { if observeErr != nil { panic(observeErr) } - return Snapshot{State: current, Observation: observation, Effects: domain.effectCounts(), Receipts: receipts.snapshot(), CommitCount: commits} + return Snapshot{State: current, Observation: observation, Effects: domain.effectCounts(), Receipts: store.instanceReceipts().snapshot(), CommitCount: commits} }, } fixture.New = func(_ testing.TB, requested Setup) KernelConformance { diff --git a/boatstack/kernel/conformance/lifecycle.go b/boatstack/kernel/conformance/lifecycle.go index ebc984f..e3391ea 100644 --- a/boatstack/kernel/conformance/lifecycle.go +++ b/boatstack/kernel/conformance/lifecycle.go @@ -519,7 +519,9 @@ func (suite ObjectiveLifecycleConformance) failedAttemptsNeverAdvance(t *testing } } -// replay proves settled and cross-instance prescriptions cannot re-commit. +// replay proves a settled prescription reconciles to its original durable +// receipt without re-committing, and cross-instance prescriptions cannot +// commit at all. func (suite ObjectiveLifecycleConformance) replay(t *testing.T) { fixture := suite.fresh(t) base := stageLifecycleValue(t, fixture, "root", 1, "v1") @@ -527,15 +529,20 @@ func (suite ObjectiveLifecycleConformance) replay(t *testing.T) { runtime := fixture.Reopen(t) applyLifecycle(t, runtime, fixture, fixture.BindTransition, &base) request, prescription := resolveLifecycle(t, runtime, fixture, fixture.AdvanceTransition, &next) - if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}); err != nil { + original, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { t.Fatal(err) } before := fixture.Snapshot() - if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}); err == nil { - t.Fatal("settled lifecycle prescription replayed") + reconciled, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { + t.Fatalf("control-law objective-lifecycle-reconciliation: settled retry = %v, want the original durable receipt", err) + } + if !reflect.DeepEqual(reconciled, original) { + t.Fatalf("control-law objective-lifecycle-reconciliation: retry returned a different receipt: original=%#v reconciled=%#v", original, reconciled) } if after := fixture.Snapshot(); !reflect.DeepEqual(before, after) { - t.Fatalf("control-law objective-lifecycle-replay: replay changed accepted facts: before=%#v after=%#v", before, after) + t.Fatalf("control-law objective-lifecycle-replay: reconciliation changed accepted facts: before=%#v after=%#v", before, after) } t.Run("cross instance", func(t *testing.T) { diff --git a/boatstack/kernel/conformance/lifecycle_law_test.go b/boatstack/kernel/conformance/lifecycle_law_test.go index 9dad729..0b8de48 100644 --- a/boatstack/kernel/conformance/lifecycle_law_test.go +++ b/boatstack/kernel/conformance/lifecycle_law_test.go @@ -80,22 +80,27 @@ func TestLifecycleSuiteRejectsBlindConcurrentSuccessorCommit(t *testing.T) { // the prior binding, resurrecting cleared state. type resurrectingLifecycleStore struct{ inner *registerStore } -func (s resurrectingLifecycleStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { +func (s resurrectingLifecycleStore) Create(ctx context.Context, instanceID string, initial kernel.ControlState) error { + return s.inner.Create(ctx, instanceID, initial) +} + +func (s resurrectingLifecycleStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { return s.inner.Load(ctx, instanceID) } -func (s resurrectingLifecycleStore) BeginEffect(ctx context.Context, revision uint64, target kernel.ControlState) error { - return s.inner.BeginEffect(ctx, revision, target) +func (s resurrectingLifecycleStore) BeginEffect(ctx context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { + return s.inner.BeginEffect(ctx, instanceID, revision, attempt) } -func (s resurrectingLifecycleStore) CommitTransition(ctx context.Context, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { +func (s resurrectingLifecycleStore) CommitTransition(ctx context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { s.inner.mu.Lock() - if target.ObjectiveBinding == nil && s.inner.state.ObjectiveBinding != nil { - binding := *s.inner.state.ObjectiveBinding + instance := s.inner.instances[instanceID] + if target.ObjectiveBinding == nil && instance.state.ObjectiveBinding != nil { + binding := *instance.state.ObjectiveBinding target.ObjectiveBinding = &binding } s.inner.mu.Unlock() - return s.inner.CommitTransition(ctx, revision, target, receipt) + return s.inner.CommitTransition(ctx, instanceID, revision, target, receipt) } func TestLifecycleSuiteRejectsClearedBindingResurrection(t *testing.T) { diff --git a/boatstack/kernel/conformance/lifecycle_register.go b/boatstack/kernel/conformance/lifecycle_register.go index d407b9b..19ebea0 100644 --- a/boatstack/kernel/conformance/lifecycle_register.go +++ b/boatstack/kernel/conformance/lifecycle_register.go @@ -314,7 +314,7 @@ func newLifecycleRegisterFixture() (ObjectiveLifecycleConformance, lifecyclePort domain := &lifecycleDomain{candidates: candidates, verifications: map[string]int{}} operator := &lifecycleOperator{candidates: candidates, executions: map[string]int{}} initial := kernel.ControlState{InstanceID: instanceID, Program: program.Identity(), Mode: "supervising", Revision: 1} - store := ®isterStore{state: cloneSettlementState(initial)} + store := newRegisterStore(initial) clock := settlementClock{now: now} classifier := lifecycleCapabilities{} diff --git a/boatstack/kernel/conformance/revisioned_register.go b/boatstack/kernel/conformance/revisioned_register.go index b94b45b..163344a 100644 --- a/boatstack/kernel/conformance/revisioned_register.go +++ b/boatstack/kernel/conformance/revisioned_register.go @@ -327,50 +327,107 @@ func (registerCapabilities) RequiredCapabilities(transition kernel.Transition) ( } } +// registerInstance is one durable instance record in the settlement store. +type registerInstance struct { + state kernel.ControlState + receipts []kernel.Receipt + commits int +} + +// registerStore is the keyed multi-instance settlement Store: state, +// receipts, revisions, and compare-and-swap are local to one instance. type registerStore struct { mu sync.Mutex - state kernel.ControlState - receipts []kernel.Receipt + instances map[string]*registerInstance + current string commitFailures int - commitCount int } -func (s *registerStore) Load(context.Context, string) (kernel.ControlState, error) { +// newRegisterStore returns a settlement store provisioned with one initial +// instance record. +func newRegisterStore(initial kernel.ControlState) *registerStore { + store := ®isterStore{instances: map[string]*registerInstance{}} + if err := store.Create(context.Background(), initial.InstanceID, initial); err != nil { + panic(err) + } + return store +} + +func (s *registerStore) Create(_ context.Context, instanceID string, initial kernel.ControlState) error { s.mu.Lock() defer s.mu.Unlock() - return cloneSettlementState(s.state), nil + if initial.InstanceID != instanceID { + return fmt.Errorf("initial control state belongs to %q, not %q", initial.InstanceID, instanceID) + } + if _, ok := s.instances[instanceID]; ok { + return kernel.InstanceExistsError{InstanceID: instanceID} + } + s.instances[instanceID] = ®isterInstance{state: cloneSettlementState(initial)} + if s.current == "" { + s.current = instanceID + } + return nil +} + +func (s *registerStore) Load(_ context.Context, instanceID string) (kernel.InstanceRecord, error) { + s.mu.Lock() + defer s.mu.Unlock() + instance, ok := s.instances[instanceID] + if !ok { + return kernel.InstanceRecord{}, kernel.InstanceNotFoundError{InstanceID: instanceID} + } + return kernel.InstanceRecord{State: cloneSettlementState(instance.state), Receipts: append([]kernel.Receipt(nil), instance.receipts...)}, nil } -func (s *registerStore) BeginEffect(_ context.Context, revision uint64, target kernel.ControlState) error { +func (s *registerStore) BeginEffect(_ context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { s.mu.Lock() defer s.mu.Unlock() - if s.state.Revision != revision { + instance, ok := s.instances[instanceID] + if !ok { + return kernel.InstanceNotFoundError{InstanceID: instanceID} + } + if instance.state.Revision != revision { return fmt.Errorf("stale revision") } - s.state = cloneSettlementState(target) + instance.state = cloneSettlementState(attempt) return nil } -func (s *registerStore) CommitTransition(_ context.Context, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { +func (s *registerStore) CommitTransition(_ context.Context, instanceID string, revision uint64, target kernel.ControlState, receipt kernel.Receipt) error { s.mu.Lock() defer s.mu.Unlock() + instance, ok := s.instances[instanceID] + if !ok { + return kernel.InstanceNotFoundError{InstanceID: instanceID} + } if s.commitFailures > 0 { s.commitFailures-- return fmt.Errorf("simulated atomic transaction failure") } - if s.state.Revision != revision { + if instance.state.Revision != revision { return fmt.Errorf("stale revision") } - s.state = cloneSettlementState(target) - s.receipts = append(s.receipts, receipt) - s.commitCount++ + instance.state = cloneSettlementState(target) + instance.receipts = append(instance.receipts, receipt) + instance.commits++ return nil } +// currentInstance returns the fixture's active instance record. Fixture +// hooks act on this record; the kernel-facing Store methods never use it. +func (s *registerStore) currentInstance() *registerInstance { + instance, ok := s.instances[s.current] + if !ok { + panic(fmt.Sprintf("register store has no current instance %q", s.current)) + } + return instance +} + func (s *registerStore) snapshot() (kernel.ControlState, []kernel.Receipt, int) { s.mu.Lock() defer s.mu.Unlock() - return cloneSettlementState(s.state), append([]kernel.Receipt(nil), s.receipts...), s.commitCount + instance := s.currentInstance() + return cloneSettlementState(instance.state), append([]kernel.Receipt(nil), instance.receipts...), instance.commits } func (s *registerStore) failNextCommit() { @@ -382,19 +439,25 @@ func (s *registerStore) failNextCommit() { func (s *registerStore) bumpRevision() { s.mu.Lock() defer s.mu.Unlock() - s.state.Revision++ + s.currentInstance().state.Revision++ } func (s *registerStore) retargetProgram(program kernel.ProgramIdentity) { s.mu.Lock() defer s.mu.Unlock() - s.state.Program = program + s.currentInstance().state.Program = program } +// retargetInstance provisions a separate control instance whose state +// mirrors the active one and makes it the fixture's active instance, so laws +// can address a prescription minted for one instance against another. func (s *registerStore) retargetInstance(instanceID string) { s.mu.Lock() defer s.mu.Unlock() - s.state.InstanceID = instanceID + state := cloneSettlementState(s.currentInstance().state) + state.InstanceID = instanceID + s.instances[instanceID] = ®isterInstance{state: state} + s.current = instanceID } func (s *registerStore) setBinding(objective kernel.Objective) { @@ -404,28 +467,48 @@ func (s *registerStore) setBinding(objective kernel.Objective) { if err != nil { panic(err) } - s.state.ObjectiveBinding = &binding + s.currentInstance().state.ObjectiveBinding = &binding } func (s *registerStore) removeReceipts() { s.mu.Lock() defer s.mu.Unlock() - s.receipts = nil + s.currentInstance().receipts = nil } func (s *registerStore) substituteLastReceipt(receipt kernel.Receipt) { s.mu.Lock() defer s.mu.Unlock() - if len(s.receipts) == 0 { + instance := s.currentInstance() + if len(instance.receipts) == 0 { panic("no committed receipt to substitute") } - s.receipts[len(s.receipts)-1] = receipt + instance.receipts[len(instance.receipts)-1] = receipt } func (s *registerStore) reset(state kernel.ControlState) { s.mu.Lock() defer s.mu.Unlock() - s.state = cloneSettlementState(state) + s.currentInstance().state = cloneSettlementState(state) +} + +// forceState overwrites the requested instance's state without any +// compare-and-swap, for dishonest white-box store variants. +func (s *registerStore) forceState(instanceID string, state kernel.ControlState) { + s.mu.Lock() + defer s.mu.Unlock() + s.instances[instanceID].state = cloneSettlementState(state) +} + +// forceCommit commits state plus receipt without any compare-and-swap, for +// dishonest white-box store variants. +func (s *registerStore) forceCommit(instanceID string, target kernel.ControlState, receipt kernel.Receipt) { + s.mu.Lock() + defer s.mu.Unlock() + instance := s.instances[instanceID] + instance.state = cloneSettlementState(target) + instance.receipts = append(instance.receipts, receipt) + instance.commits++ } type settlementLocker struct{ mu sync.Mutex } @@ -484,9 +567,9 @@ func newRevisionedRegisterFixture() (SettlementConformance, registerPorts) { candidates := newRegisterCandidates() domain := ®isterDomain{candidates: candidates, verifications: map[string]int{}} operator := ®isterOperator{candidates: candidates, executions: map[string]int{}} - store := ®isterStore{state: kernel.ControlState{ + store := newRegisterStore(kernel.ControlState{ InstanceID: instanceID, Program: program.Identity(), Mode: "unbound", Revision: 1, - }} + }) clock := settlementClock{now: now} classifier := registerCapabilities{} diff --git a/boatstack/kernel/conformance/settlement.go b/boatstack/kernel/conformance/settlement.go index 312d867..ca8e243 100644 --- a/boatstack/kernel/conformance/settlement.go +++ b/boatstack/kernel/conformance/settlement.go @@ -297,7 +297,8 @@ func (suite SettlementConformance) replay(t *testing.T) { objective := stageRegisterValue(t, fixture, "alpha") runtime := fixture.Reopen(t) request, prescription := resolveSettlement(t, runtime, fixture, objective) - if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}); err != nil { + original, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { t.Fatal(err) } before := fixture.Snapshot() @@ -305,8 +306,12 @@ func (suite SettlementConformance) replay(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}); err == nil { - t.Fatal("settled prescription replayed") + reconciled, err := runtime.Apply(context.Background(), kernel.ApplyRequest{ResolveRequest: request, Prescription: prescription}) + if err != nil { + t.Fatalf("control-law trusted-settlement-reconciliation: settled retry = %v, want the original durable receipt", err) + } + if !reflect.DeepEqual(reconciled, original) { + t.Fatalf("control-law trusted-settlement-reconciliation: retry returned a different receipt: original=%#v reconciled=%#v", original, reconciled) } after := fixture.Snapshot() afterAccepted, err := fixture.Accepted() @@ -314,7 +319,7 @@ func (suite SettlementConformance) replay(t *testing.T) { t.Fatal(err) } if !reflect.DeepEqual(before, after) || !reflect.DeepEqual(beforeAccepted, afterAccepted) { - t.Fatalf("control-law trusted-settlement-replay: replay changed accepted facts: before=%#v after=%#v", before, after) + t.Fatalf("control-law trusted-settlement-replay: reconciliation changed accepted facts: before=%#v after=%#v", before, after) } } diff --git a/boatstack/kernel/conformance/settlement_law_test.go b/boatstack/kernel/conformance/settlement_law_test.go index 374e222..17254bd 100644 --- a/boatstack/kernel/conformance/settlement_law_test.go +++ b/boatstack/kernel/conformance/settlement_law_test.go @@ -103,22 +103,27 @@ func TestSettlementSuiteRejectsLatestStagedReader(t *testing.T) { // tornSettlementStore persists the target state but silently drops the receipt. type tornSettlementStore struct{ inner *registerStore } -func (s tornSettlementStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { +func (s tornSettlementStore) Create(ctx context.Context, instanceID string, initial kernel.ControlState) error { + return s.inner.Create(ctx, instanceID, initial) +} + +func (s tornSettlementStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { return s.inner.Load(ctx, instanceID) } -func (s tornSettlementStore) BeginEffect(ctx context.Context, revision uint64, target kernel.ControlState) error { - return s.inner.BeginEffect(ctx, revision, target) +func (s tornSettlementStore) BeginEffect(ctx context.Context, instanceID string, revision uint64, attempt kernel.ControlState) error { + return s.inner.BeginEffect(ctx, instanceID, revision, attempt) } -func (s tornSettlementStore) CommitTransition(_ context.Context, revision uint64, target kernel.ControlState, _ kernel.Receipt) error { +func (s tornSettlementStore) CommitTransition(_ context.Context, instanceID string, revision uint64, target kernel.ControlState, _ kernel.Receipt) error { s.inner.mu.Lock() defer s.inner.mu.Unlock() - if s.inner.state.Revision != revision { + instance := s.inner.instances[instanceID] + if instance.state.Revision != revision { return fmt.Errorf("stale revision") } - s.inner.state = cloneSettlementState(target) - s.inner.commitCount++ + instance.state = cloneSettlementState(target) + instance.commits++ return nil } @@ -145,24 +150,22 @@ type blindSettlementStore struct { barrier *twoPartyBarrier } -func (s blindSettlementStore) Load(ctx context.Context, instanceID string) (kernel.ControlState, error) { +func (s blindSettlementStore) Create(ctx context.Context, instanceID string, initial kernel.ControlState) error { + return s.inner.Create(ctx, instanceID, initial) +} + +func (s blindSettlementStore) Load(ctx context.Context, instanceID string) (kernel.InstanceRecord, error) { return s.inner.Load(ctx, instanceID) } -func (s blindSettlementStore) BeginEffect(_ context.Context, _ uint64, target kernel.ControlState) error { +func (s blindSettlementStore) BeginEffect(_ context.Context, instanceID string, _ uint64, attempt kernel.ControlState) error { _ = s.barrier.wait() - s.inner.mu.Lock() - defer s.inner.mu.Unlock() - s.inner.state = cloneSettlementState(target) + s.inner.forceState(instanceID, attempt) return nil } -func (s blindSettlementStore) CommitTransition(_ context.Context, _ uint64, target kernel.ControlState, receipt kernel.Receipt) error { - s.inner.mu.Lock() - defer s.inner.mu.Unlock() - s.inner.state = cloneSettlementState(target) - s.inner.receipts = append(s.inner.receipts, receipt) - s.inner.commitCount++ +func (s blindSettlementStore) CommitTransition(_ context.Context, instanceID string, _ uint64, target kernel.ControlState, receipt kernel.Receipt) error { + s.inner.forceCommit(instanceID, target, receipt) return nil } diff --git a/boatstack/kernel/instance.go b/boatstack/kernel/instance.go new file mode 100644 index 0000000..c0fd7aa --- /dev/null +++ b/boatstack/kernel/instance.go @@ -0,0 +1,98 @@ +package kernel + +import ( + "errors" + "fmt" +) + +// InstanceRecord is the complete durable record of one control instance: the +// current ControlState plus the ordered committed Receipt history. A Store +// must load both atomically so a caller can never observe a target state +// without the receipt that committed it. +type InstanceRecord struct { + State ControlState `json:"state"` + Receipts []Receipt `json:"receipts"` +} + +// InstanceNotFoundError is the typed result for a control instance that has +// not been provisioned. The kernel never manufactures objective state for a +// missing instance. +type InstanceNotFoundError struct{ InstanceID string } + +func (e InstanceNotFoundError) Error() string { + return fmt.Sprintf("control instance %q does not exist", e.InstanceID) +} + +// IsInstanceNotFound reports whether err carries a typed instance-not-found +// result. +func IsInstanceNotFound(err error) bool { + var target InstanceNotFoundError + return errors.As(err, &target) +} + +// InstanceExistsError is the typed result for an atomic creation that would +// overwrite an existing instance record. +type InstanceExistsError struct{ InstanceID string } + +func (e InstanceExistsError) Error() string { + return fmt.Sprintf("control instance %q already exists", e.InstanceID) +} + +// IsInstanceExists reports whether err carries a typed instance-exists result. +func IsInstanceExists(err error) bool { + var target InstanceExistsError + return errors.As(err, &target) +} + +// Validate fails closed on a loaded instance record whose durable evidence +// cannot be trusted: malformed or fabricated receipts (content identity), +// receipts routed from another instance, duplicate settlement of one +// prescription, rewound or reordered receipt revisions, objective lineage +// that does not extend the committed chain, and a control state that has +// fallen behind its own committed history. +// +// It deliberately allows two legitimate shapes: a recovery-required state +// whose persisted attempt has no final receipt, and a state revision ahead +// of the committed history (an external writer may advance an instance; +// prescription freshness, not record validation, decides staleness). +func (r InstanceRecord) Validate(instanceID string) error { + if err := r.State.Validate(); err != nil { + return fmt.Errorf("instance record control state is invalid: %w", err) + } + if r.State.InstanceID != instanceID { + return fmt.Errorf("instance record control state belongs to %q, not requested instance %q", r.State.InstanceID, instanceID) + } + var lastResult uint64 + var chain *ObjectiveBinding + settled := make(map[string]bool, len(r.Receipts)) + for index, receipt := range r.Receipts { + if err := receipt.Validate(); err != nil { + return fmt.Errorf("committed receipt %d is invalid: %w", index, err) + } + if receipt.InstanceID != instanceID { + return fmt.Errorf("committed receipt %q belongs to instance %q, not %q", receipt.ID, receipt.InstanceID, instanceID) + } + if settled[receipt.PrescriptionID] { + return fmt.Errorf("prescription %q is settled by more than one committed receipt", receipt.PrescriptionID) + } + settled[receipt.PrescriptionID] = true + if receipt.PriorStateRevision < lastResult { + return fmt.Errorf("committed receipt %q rewinds or reorders the state revision sequence", receipt.ID) + } + lastResult = receipt.ResultStateRevision + if receipt.ObjectiveMutation == PreserveObjective { + continue + } + if !equalBinding(receipt.PriorObjectiveBinding, chain) { + return fmt.Errorf("committed receipt %q does not extend the committed objective lineage", receipt.ID) + } + chain = receipt.ResultObjectiveBinding + } + if r.State.Revision < lastResult { + return fmt.Errorf("control state revision %d is behind the committed receipt history at %d", r.State.Revision, lastResult) + } + if r.State.Recovery != nil && r.State.Revision == lastResult { + return fmt.Errorf("recovery state requires a persisted attempt beyond the committed history") + } + return nil +} diff --git a/boatstack/kernel/runtime.go b/boatstack/kernel/runtime.go index 2809b74..476f67a 100644 --- a/boatstack/kernel/runtime.go +++ b/boatstack/kernel/runtime.go @@ -69,14 +69,26 @@ type CapabilityClassifier interface { RequiredCapabilities(Transition) ([]Capability, error) } -// Store owns the durable transaction boundary. BeginEffect must atomically -// persist an unresolved attempt before an operator can run. CommitTransition -// must atomically replace that attempt with the target state and its receipt: -// either both become visible or neither does. +// Store owns the durable multi-instance persistence boundary. Every method +// names its control instance explicitly; operations on one instance must not +// load, mutate, or append history belonging to another. +// +// Create must atomically provision a new instance record and must return a +// typed InstanceExistsError instead of overwriting an existing record, even +// under concurrent creation. Load must atomically return the current +// ControlState together with the complete ordered committed Receipt history, +// and must return a typed InstanceNotFoundError for a missing instance. +// BeginEffect must atomically persist an unresolved attempt, guarded by a +// per-instance revision compare-and-swap, before an operator can run. +// CommitTransition must atomically replace that attempt with the target +// state plus exactly one appended receipt — either both become visible or +// neither does — under the same per-instance compare-and-swap. Receipt +// history is append-only. type Store interface { - Load(context.Context, string) (ControlState, error) - BeginEffect(context.Context, uint64, ControlState) error - CommitTransition(context.Context, uint64, ControlState, Receipt) error + Create(ctx context.Context, instanceID string, initial ControlState) error + Load(ctx context.Context, instanceID string) (InstanceRecord, error) + BeginEffect(ctx context.Context, instanceID string, expectedRevision uint64, attempt ControlState) error + CommitTransition(ctx context.Context, instanceID string, expectedRevision uint64, target ControlState, receipt Receipt) error } type Lock interface{ Unlock() error } @@ -185,16 +197,40 @@ func NewRuntime(program Program, domain Domain, operator Operator, classifier Ca return Runtime{program: program, domain: domain, operator: operator, classifier: classifier, store: store, locker: locker, clock: clock}, nil } +// Provision atomically creates a durable control instance at the program's +// declared initial mode. It never manufactures accepted objective state: the +// initial ControlState carries no objective binding and no recovery. A +// missing instance stays a typed InstanceNotFoundError until an operator +// provisions it explicitly; duplicate or concurrent provisioning surfaces +// the store's typed InstanceExistsError and leaves the existing record +// untouched. +func (r Runtime) Provision(ctx context.Context, instanceID string) (ControlState, error) { + if !semanticID.MatchString(instanceID) { + return ControlState{}, fmt.Errorf("control instance identity %q is not a semantic identifier", instanceID) + } + initial := ControlState{InstanceID: instanceID, Program: r.program.Identity(), Mode: r.program.InitialMode, Revision: 1} + if err := initial.Validate(); err != nil { + return ControlState{}, fmt.Errorf("initial control state is invalid: %w", err) + } + if err := r.store.Create(ctx, instanceID, initial); err != nil { + return ControlState{}, err + } + return initial, nil +} + func (r Runtime) Resolve(ctx context.Context, request ResolveRequest) (Resolution, error) { - state, err := r.store.Load(ctx, request.InstanceID) + record, err := r.store.Load(ctx, request.InstanceID) if err != nil { return Resolution{}, err } + if err := record.Validate(request.InstanceID); err != nil { + return Resolution{}, fmt.Errorf("instance record failed fail-closed validation: %w", err) + } observation, err := r.domain.Observe(ctx, request.InstanceID) if err != nil { return Resolution{}, err } - return r.resolve(ctx, state, observation, request) + return r.resolve(ctx, record.State, observation, request) } func (r Runtime) resolve(ctx context.Context, state ControlState, observation Observation, request ResolveRequest) (Resolution, error) { @@ -369,10 +405,19 @@ func (r Runtime) Apply(ctx context.Context, request ApplyRequest) (Receipt, erro return Receipt{}, err } defer lock.Unlock() - state, err := r.store.Load(ctx, request.InstanceID) + record, err := r.store.Load(ctx, request.InstanceID) if err != nil { return Receipt{}, err } + if err := record.Validate(request.InstanceID); err != nil { + return Receipt{}, fmt.Errorf("instance record failed fail-closed validation: %w", err) + } + state := record.State + if receipt, reconciled, err := reconcileCommittedResult(record, request); err != nil { + return Receipt{}, err + } else if reconciled { + return receipt, nil + } observation, err := r.domain.Observe(ctx, request.InstanceID) if err != nil { return Receipt{}, err @@ -418,7 +463,7 @@ func (r Runtime) Apply(ctx context.Context, request ApplyRequest) (Receipt, erro if err := attempt.Validate(); err != nil { return Receipt{}, fmt.Errorf("effect attempt state is invalid: %w", err) } - if err := r.store.BeginEffect(ctx, state.Revision, attempt); err != nil { + if err := r.store.BeginEffect(ctx, request.InstanceID, state.Revision, attempt); err != nil { return Receipt{}, fmt.Errorf("effect attempt did not begin: %w", err) } effect, err := r.operator.Execute(ctx, Operation{InstanceID: request.InstanceID, Transition: transition, Observation: observation, Objective: objective, Capabilities: required}) @@ -479,12 +524,84 @@ func (r Runtime) Apply(ctx context.Context, request ApplyRequest) (Receipt, erro if err := receipt.Validate(); err != nil { return Receipt{}, RecoveryRequiredError{Reason: "transition receipt is invalid: " + err.Error()} } - if err := r.store.CommitTransition(ctx, attempt.Revision, target, receipt); err != nil { + if err := r.store.CommitTransition(ctx, request.InstanceID, attempt.Revision, target, receipt); err != nil { return Receipt{}, RecoveryRequiredError{Reason: "state and receipt transaction did not commit: " + err.Error()} } return receipt, nil } +// reconcileCommittedResult recovers the durable outcome of an Apply whose +// exact transition already committed but whose response was lost. The +// content-addressed Prescription ID is the committed request identity: when +// exactly one valid committed receipt settles it, that original receipt is +// returned without observing, executing, verifying, mutating state, +// appending a receipt, or advancing a revision. A receipt that does not +// match the prescription exactly — substituted, misrouted, or otherwise +// divergent — refuses reconciliation instead of reporting success. A failed +// or interrupted attempt has no committed receipt and therefore never +// reconciles; it remains recovery-required on the normal path. +func reconcileCommittedResult(record InstanceRecord, request ApplyRequest) (Receipt, bool, error) { + prescription := request.Prescription + var match *Receipt + for index := range record.Receipts { + if record.Receipts[index].PrescriptionID != prescription.ID { + continue + } + if match != nil { + return Receipt{}, false, fmt.Errorf("prescription %q is settled by more than one committed receipt", prescription.ID) + } + match = &record.Receipts[index] + } + if match == nil { + if recovery := record.State.Recovery; recovery != nil && recovery.PrescriptionID == prescription.ID { + return Receipt{}, false, RecoveryRequiredError{Reason: "the attempt for this prescription is unresolved; no committed receipt settles it"} + } + return Receipt{}, false, nil + } + if err := prescription.validateIdentity(); err != nil { + return Receipt{}, false, err + } + receipt := *match + mismatch := func(field string) error { + return fmt.Errorf("committed receipt %q settles prescription %q but its %s does not match; reconciliation refused", receipt.ID, prescription.ID, field) + } + switch { + case receipt.InstanceID != request.InstanceID || receipt.InstanceID != prescription.ExpectedInstanceID: + return Receipt{}, false, mismatch("instance identity") + case receipt.TransitionID != prescription.TransitionID: + return Receipt{}, false, mismatch("transition identity") + case receipt.ObjectiveMutation != prescription.ObjectiveMutation: + return Receipt{}, false, mismatch("objective mutation") + case receipt.PriorStateRevision != prescription.ExpectedStateRevision: + return Receipt{}, false, mismatch("prior state revision") + case receipt.Program.Fingerprint != prescription.ExpectedProgramFingerprint: + return Receipt{}, false, mismatch("program fingerprint") + case receipt.PriorObservation != prescription.ExpectedSnapshotFingerprint: + return Receipt{}, false, mismatch("prior observation") + case receipt.AuthorityFingerprint != prescription.AuthorityFingerprint: + return Receipt{}, false, mismatch("authority fingerprint") + case !equalBinding(receipt.PriorObjectiveBinding, prescription.ExpectedObjectiveBinding): + return Receipt{}, false, mismatch("prior objective binding") + case !equalBinding(receipt.RequestedObjectiveBinding, prescription.RequestedObjectiveBinding): + return Receipt{}, false, mismatch("requested objective binding") + case !equalCapabilities(receipt.Capabilities, prescription.RequiredCapabilities): + return Receipt{}, false, mismatch("required capabilities") + } + return receipt, true, nil +} + +func equalCapabilities(left, right []Capability) bool { + if len(left) != len(right) { + return false + } + for index := range left { + if left[index] != right[index] { + return false + } + } + return true +} + func (r Receipt) Validate() error { identity := r want := identity.ID @@ -610,7 +727,10 @@ func newPrescription(state ControlState, observation Observation, transition Tra return p, nil } -func (p Prescription) validateCurrent(state ControlState, observation Observation, authority authorityProjection) error { +// validateIdentity proves the prescription's content-addressed identity so a +// fabricated or altered prescription can never be used to look up or replay +// a committed result. +func (p Prescription) validateIdentity() error { identity := p want := identity.ID identity.ID = "" @@ -618,6 +738,13 @@ func (p Prescription) validateCurrent(state ControlState, observation Observatio if err != nil || want != "prx-"+got { return StalePrescriptionError{Reason: "prescription content identity is invalid"} } + return nil +} + +func (p Prescription) validateCurrent(state ControlState, observation Observation, authority authorityProjection) error { + if err := p.validateIdentity(); err != nil { + return err + } bindingFingerprint, bindingErr := Fingerprint(state.ObjectiveBinding) current, freshnessErr := NewFreshness(state.InstanceID, state.Revision, state.Program.Fingerprint, observation.Fingerprint, bindingFingerprint, authority.Fingerprint) if p.SchemaVersion != PrescriptionSchemaVersion || !p.ObjectiveMutation.valid() || bindingErr != nil || freshnessErr != nil || p.Freshness.Check(current) != nil || !equalBinding(p.ExpectedObjectiveBinding, state.ObjectiveBinding) { diff --git a/docs/architecture/kernel.md b/docs/architecture/kernel.md index 994442d..c1d029c 100644 --- a/docs/architecture/kernel.md +++ b/docs/architecture/kernel.md @@ -54,7 +54,10 @@ The general kernel never imports the software-delivery implementation. - operator-neutral execution and fresh postcondition verification; - program-defined marked modes; - explicit recovery state and recovery transitions; -- domain-neutral committed receipts. +- domain-neutral committed receipts; +- explicit instance provisioning and atomic instance records: current control + state plus the complete ordered committed receipt history; +- idempotent committed-result reconciliation by exact prescription identity. ## Software-delivery-owned semantics @@ -133,7 +136,10 @@ Programs declare capabilities, but a trusted capability classifier supplies the minimum for each concrete operation. The operator receives only that admitted set. Effect facts must stay inside transition-owned facets. -The generic `Store` is one durability boundary. `BeginEffect` atomically +The generic `Store` is one durability boundary. `Create` atomically +provisions a new instance record and cannot overwrite an existing one. +`Load` returns the atomic instance record: current control state plus the +complete ordered committed receipt history. `BeginEffect` atomically persists the attempt state — including its recovery obligation — before any operator effect. `CommitTransition` atomically persists the target control state and its verified receipt; neither may become visible alone. If an @@ -176,6 +182,59 @@ missing or substituted receipt fails closed rather than inventing accepted content. Conformance detects dishonest store or reader implementations; the runtime does not claim to make adversarial ports safe. +## Durable control instances + +One store holds many durable control instances. The control law is stated +over the persistence boundary itself, not over any one caller: + +> Every persistence operation — create, load, begin-effect, commit — and +> every lock acquisition names its exact control instance. State, revision +> compare-and-swap, recovery, locking, and committed receipt history are +> local to that one instance. A loaded record exposes the current control +> state and its complete ordered committed history atomically, and fails +> closed when that history cannot be trusted. Retrying an `Apply` whose +> exact prescription already committed returns the original durable receipt +> with no new observation, execution, verification, state mutation, receipt, +> or revision. A failed or interrupted attempt without a committed receipt +> is never reported as successful. + +Provisioning is explicit: `Runtime.Provision` creates the valid initial +control state for an instance atomically, concurrent creation yields exactly +one initial history, a duplicate is a typed instance-exists result, and a +missing instance is a typed not-found result. The kernel never silently +manufactures accepted objective state. + +Committed-result reconciliation uses the content-addressed prescription +identity as the committed request identity. Before observing or executing +anything, `Apply` searches the loaded history for exactly one valid receipt +that settles the same prescription on the same instance, revalidates that +receipt against the prescription's transition, lineage, revisions, +observation, and authority fingerprints, and returns it unchanged. This is +reconciliation of a proven historical commit — recovery of a lost successful +response — not permission to replay an effect. A pending recovery obligation +for the same prescription stays recovery-required. + +Every loaded record is validated fail-closed before use: malformed or +fabricated receipts, receipts routed from another instance, duplicate +settlement of one prescription, rewound or reordered revisions, lineage that +does not extend the committed chain, and control state behind its own +history are all rejected. A persisted attempt without a final receipt and a +state legitimately ahead of its history remain valid recovery shapes. + +`InstanceStoreConformance` in `boatstack/kernel/conformance` proves the law +for any `Store`: creation, restart, isolation, per-instance CAS and locking, +atomic state-plus-receipt commits, append-only history, exact committed +retry with zero side effects, failed-attempt retries staying +recovery-required, cross-instance replay rejection, concurrent retries +returning one durable result, corrupted-history fail-closure, and restart +reconstruction. White-box counterexamples prove the suite rejects a +singleton store that ignores the requested instance, torn state-receipt +commits, a blind global compare-and-swap, latest-receipt rather than +exact-prescription lookup, replaced or truncated history, and synthetic +success after an unresolved attempt. The suite runs against the integer +memory store, the settlement register store, and the reviewer's on-disk +file store. + ## Non-software proof fixtures `boatstack/kernel/runtime_test.go` runs an integer control instance: @@ -225,6 +284,13 @@ receipt substitution, and restart state reset. 17. Lifecycle explicitness: every binding mutation declares exactly one of bind-initial, advance, replace, or clear, satisfies that relation's own capability, and commits a receipt that independently proves the relation. +18. Instance isolation: state, revisions, CAS, recovery, locking, and receipt + history are local to one explicitly named control instance. +19. Explicit provisioning: instances are created atomically, never + overwritten, and never manufactured for a missing identity. +20. Committed-result reconciliation: an exact committed retry returns the + original durable receipt with zero side effects, and an uncommitted + attempt is never reported as successful. ## Current implementation anchors @@ -234,3 +300,5 @@ receipt substitution, and restart state reset. - [Domain-neutral conformance fixture](../../boatstack/kernel/conformance/integer.go) - [Settlement fixture](../../boatstack/kernel/conformance/revisioned_register.go) - [Settlement laws](../../boatstack/kernel/conformance/settlement.go) +- [Instance record and typed results](../../boatstack/kernel/instance.go) +- [Instance-store laws](../../boatstack/kernel/conformance/instance_store.go) diff --git a/release-notes/2026-08-23-durable-control-instances.md b/release-notes/2026-08-23-durable-control-instances.md new file mode 100644 index 0000000..702c368 --- /dev/null +++ b/release-notes/2026-08-23-durable-control-instances.md @@ -0,0 +1,35 @@ +### Durable, isolated, idempotently resumable control instances + +The kernel `Store` is now an explicit multi-instance persistence contract. +`Create` provisions an instance atomically and cannot overwrite an existing +record; concurrent creation yields exactly one initial history and one typed +instance-exists loser, and a missing instance is a typed not-found result +rather than manufactured state. `Load` returns the atomic instance record — +current control state plus the complete ordered committed receipt history — +and every persistence mutation and lock acquisition names its exact +instance, so revisions, compare-and-swap, recovery, locking, and history +stay local to one instance while independent instances progress +concurrently. + +Retrying an `Apply` whose exact prescription already committed now +reconciles instead of failing: the runtime finds the single valid receipt +that settles that prescription on that instance, revalidates it against the +prescription's transition, lineage, revisions, observation, and authority +fingerprints, and returns the original durable receipt with no new +observation, execution, verification, state mutation, receipt, or revision. +A failed or interrupted attempt without a committed receipt stays +recovery-required and is never reported as successful. Every loaded record +is validated fail-closed against fabricated, misrouted, duplicated, +reordered, or truncated history. + +A reusable `InstanceStoreConformance` suite proves the law for any store — +creation and restart, isolation, per-instance CAS and locking, atomic +state-plus-receipt commits, append-only history, exact committed retry with +zero side effects, cross-instance replay rejection, concurrent retries +returning one durable result, corrupted-history fail-closure, and restart +reconstruction — and its white-box counterexamples reject singleton stores, +torn commits, blind global CAS, latest-receipt lookup, replaced or truncated +history, and synthetic success. The suite runs against the integer memory +store, the settlement register store, and the reviewer's on-disk file store, +which now provisions its review instance explicitly at mutating command +boundaries and locks per instance. From ac4c02b551dfd3a5a4fec5f6c0e2f8c194431b82 Mon Sep 17 00:00:00 2001 From: bigboateng Date: Sun, 23 Aug 2026 21:08:18 +0100 Subject: [PATCH 2/3] Report unprovisioned review instances read-only from status `boatstack-reviewer status` no longer fails on a branch whose control instance has not been provisioned. It reports the deterministic initial state the first mutating command will create, marked provisioned=false, without persisting anything, so read-only observers like the self-review workflow keep working while provisioning stays explicit and mutating. --- boatstack/cmd/boatstack-reviewer/main.go | 59 +++++++++++++------ .../cmd/boatstack-reviewer/reviewer_test.go | 51 ++++++++++++++++ 2 files changed, 92 insertions(+), 18 deletions(-) diff --git a/boatstack/cmd/boatstack-reviewer/main.go b/boatstack/cmd/boatstack-reviewer/main.go index cce6472..83222d0 100644 --- a/boatstack/cmd/boatstack-reviewer/main.go +++ b/boatstack/cmd/boatstack-reviewer/main.go @@ -366,39 +366,62 @@ func commandStatus(arguments []string) error { if err != nil { return err } - record, err := loop.store.Load(context.Background(), loop.instance) - if kernel.IsInstanceNotFound(err) { - return fmt.Errorf("%w; run `boatstack-reviewer resolve` to provision it", err) - } + report, err := loop.statusReport() if err != nil { return err } + return printJSON(report) +} + +type statusReport struct { + Instance string `json:"instance"` + Program kernel.ProgramIdentity `json:"program"` + State kernel.ControlState `json:"state"` + Provisioned bool `json:"provisioned"` + ProgramStale bool `json:"program_stale"` + Observation observationValue `json:"observation"` + Guidance string `json:"guidance,omitempty"` +} + +// statusReport observes the committed control state read-only. It never +// provisions: for a missing instance it reports the deterministic initial +// state the first mutating command will create, marked unprovisioned, +// without persisting anything. +func (c *loopContext) statusReport() (statusReport, error) { + record, err := c.store.Load(context.Background(), c.instance) + provisioned := true + if kernel.IsInstanceNotFound(err) { + provisioned = false + record = kernel.InstanceRecord{State: kernel.ControlState{ + InstanceID: c.instance, Program: c.program.Identity(), + Mode: c.program.InitialMode, Revision: 1, + }} + } else if err != nil { + return statusReport{}, err + } state := record.State - observed, err := loop.domain.observeValue() + observed, err := c.domain.observeValue() if err != nil { - return err + return statusReport{}, err } - stale := state.Program != loop.program.Identity() - return printJSON(struct { - Instance string `json:"instance"` - Program kernel.ProgramIdentity `json:"program"` - State kernel.ControlState `json:"state"` - ProgramStale bool `json:"program_stale"` - Observation observationValue `json:"observation"` - Guidance string `json:"guidance,omitempty"` - }{ - Instance: loop.instance, - Program: loop.program.Identity(), + stale := state.Program != c.program.Identity() + return statusReport{ + Instance: c.instance, + Program: c.program.Identity(), State: state, + Provisioned: provisioned, ProgramStale: stale, Observation: observed, Guidance: func() string { + if !provisioned { + return "this control instance is not provisioned yet; the first mutating command (`boatstack-reviewer resolve`) provisions it" + } if stale { return "the admitted policy or law changed since this state was committed; `boatstack-reviewer reset --confirm` archives it" } return submissionGuidance(state.Mode) }(), - }) + }, nil } // commandShow prints a recorded review itself — the exact archived findings diff --git a/boatstack/cmd/boatstack-reviewer/reviewer_test.go b/boatstack/cmd/boatstack-reviewer/reviewer_test.go index 7c7c918..87e6ea0 100644 --- a/boatstack/cmd/boatstack-reviewer/reviewer_test.go +++ b/boatstack/cmd/boatstack-reviewer/reviewer_test.go @@ -1084,3 +1084,54 @@ func TestGitBinaryIsAvailableForThisSuite(t *testing.T) { t.Fatal("this test suite requires git on PATH") } } + +// control-law explicit-instance-provisioning: read-only status reports the +// deterministic initial state for an unprovisioned instance without +// persisting anything; only a mutating command provisions the instance. +func TestStatusReportsUnprovisionedInstanceWithoutPersisting(t *testing.T) { + scratch := newScratchRepo(t) + policy := testPolicy(t, scratch) + program, err := compileReviewProgram(policy) + if err != nil { + t.Fatal(err) + } + store := newFileStore(scratch.repo.GitDir, "feature") + domain := &reviewDomain{repo: scratch.repo, store: store, policy: policy, baseRef: "main"} + loop := &loopContext{ + repo: scratch.repo, + policy: policy, + program: program, + store: store, + domain: domain, + operator: reviewOperator{store: store}, + instance: "feature", + baseRef: "main", + } + report, err := loop.statusReport() + if err != nil { + t.Fatalf("status on an unprovisioned instance = %v, want a read-only report", err) + } + if report.Provisioned { + t.Fatal("unprovisioned instance reported as provisioned") + } + if report.State.InstanceID != "feature" || report.State.Mode != program.InitialMode || report.State.Revision != 1 || report.State.Program != program.Identity() { + t.Fatalf("unprovisioned status does not report the deterministic initial state: %#v", report.State) + } + if _, err := store.Load(context.Background(), "feature"); !kernel.IsInstanceNotFound(err) { + t.Fatalf("read-only status persisted the instance: %v", err) + } + runtime, err := loop.runtime() + if err != nil { + t.Fatal(err) + } + if err := loop.ensureProvisioned(runtime); err != nil { + t.Fatal(err) + } + provisioned, err := loop.statusReport() + if err != nil { + t.Fatal(err) + } + if !provisioned.Provisioned || provisioned.State.Revision != 1 { + t.Fatalf("provisioned status = %#v, want the durable initial state", provisioned) + } +} From dff0b854b8e696cfe195d3ee54991c40f322b6bb Mon Sep 17 00:00:00 2001 From: bigboateng Date: Sun, 23 Aug 2026 21:10:21 +0100 Subject: [PATCH 3/3] Seal converged self-review attestation --- .github/reviews/cursor-pr5-instance-persistence.receipt.json | 4 ++++ 1 file changed, 4 insertions(+) create mode 100644 .github/reviews/cursor-pr5-instance-persistence.receipt.json diff --git a/.github/reviews/cursor-pr5-instance-persistence.receipt.json b/.github/reviews/cursor-pr5-instance-persistence.receipt.json new file mode 100644 index 0000000..3bef499 --- /dev/null +++ b/.github/reviews/cursor-pr5-instance-persistence.receipt.json @@ -0,0 +1,4 @@ +{ + "reviewed_tree": "975232189c6c9be401c457c38fa72e489d6a9b53", + "program_fingerprint": "3ca3397ff275d89bdb6d5c934b86b51d3cbdfab0ee628c47fe94d1d4f5767155" +}