diff --git a/unroll/registry.go b/unroll/registry.go index 5b5584f75..bb108f63e 100644 --- a/unroll/registry.go +++ b/unroll/registry.go @@ -135,12 +135,34 @@ func (a *UnrollRegistryActor) Ref() actor.ActorRef[RegistryMsg, RegistryResp] { } // RestoreNonTerminal resumes all non-terminal records from the control store. +// +// The actual restore runs inside the registry actor's goroutine via a +// restoreNonTerminalMsg, so all mutations of r.active and r.pending stay +// serialized with concurrent Receive turns (handleEnsure / handleGetStatus +// can already be running by the time the daemon boot path reaches this +// call, because NewUnrollRegistryActor has already Start()ed the actor). func (a *UnrollRegistryActor) RestoreNonTerminal(ctx context.Context) error { - if a == nil || a.behavior == nil { + if a == nil || a.ref == nil { return fmt.Errorf("registry actor not initialized") } - return a.behavior.restoreNonTerminal(ctx) + resp, err := a.ref.Ask( + ctx, &restoreNonTerminalMsg{}, + ).Await(ctx).Unpack() + if err != nil { + return err + } + + result, ok := resp.(*restoreNonTerminalResp) + if !ok { + return fmt.Errorf("unexpected restore response %T", resp) + } + + if result.Err != "" { + return fmt.Errorf("%s", result.Err) + } + + return nil } // Stop stops the underlying registry actor. @@ -237,6 +259,40 @@ func (m *persistRecordResultMsg) MessageType() string { // registryMsgSealed seals persistRecordResultMsg into the registry surface. func (m *persistRecordResultMsg) registryMsgSealed() {} +// restoreNonTerminalMsg drives the boot-time restore of every non-terminal +// record through the registry actor's goroutine. Sending it via Ask keeps +// all mutations of r.active and r.pending serialized with concurrent +// Receive turns (handleEnsure / handleGetStatus can already be running by +// the time the daemon boot path issues this call). +type restoreNonTerminalMsg struct { + actor.BaseMessage +} + +// MessageType returns the stable message type identifier. +func (m *restoreNonTerminalMsg) MessageType() string { + return "restoreNonTerminalMsg" +} + +// registryMsgSealed seals restoreNonTerminalMsg into the registry surface. +func (m *restoreNonTerminalMsg) registryMsgSealed() {} + +// restoreNonTerminalResp carries the outcome of a boot-time restore back +// to the caller of UnrollRegistryActor.RestoreNonTerminal. +type restoreNonTerminalResp struct { + actor.BaseMessage + + // Err is populated when the restore returned an error. + Err string +} + +// MessageType returns the stable message type identifier. +func (m *restoreNonTerminalResp) MessageType() string { + return "restoreNonTerminalResp" +} + +// registryRespSealed seals restoreNonTerminalResp into the registry surface. +func (m *restoreNonTerminalResp) registryRespSealed() {} + // Receive processes one registry message. func (r *registryBehavior) Receive(ctx context.Context, msg RegistryMsg) fn.Result[RegistryResp] { @@ -257,6 +313,9 @@ func (r *registryBehavior) Receive(ctx context.Context, case *persistRecordResultMsg: return r.handlePersistRecordResult(ctx, req) + case *restoreNonTerminalMsg: + return r.handleRestoreNonTerminal(ctx) + default: return fn.Err[RegistryResp]( fmt.Errorf("unknown registry message: %T", msg), @@ -332,6 +391,54 @@ func (r *registryBehavior) handleEnsure(ctx context.Context, ) } if existing != nil { + // A durable record exists but no child is live for it. Two + // sub-cases: + // + // 1. Terminal record (Completed/Failed) — return the + // historical ActorID so callers see a stable identity + // and do not clobber the recorded sweep txid or + // failure reason. + // + // 2. Non-terminal record — the actor was admitted in a + // previous boot but never resumed (e.g. RestoreNonTerminal + // hit a transient ChainSource error). Attempt an inline + // restore so a fresh Ensure from the chain resolver or + // RPC layer can recover from a transient failure on the + // previous boot. If restore fails again, surface the + // error so the caller can retry; the durable record + // stays non-terminal and will be retried on the next + // Ensure / next daemon restart. + if !existing.IsTerminal() { + height, err := r.queryBestHeight(ctx) + if err != nil { + return fn.Err[RegistryResp]( + fmt.Errorf("best height for "+ + "restore: %w", err), + ) + } + + child, err := r.tryRestoreOne(ctx, *existing, height) + if err != nil { + return fn.Err[RegistryResp]( + fmt.Errorf("restore existing "+ + "record: %w", err), + ) + } + + r.active[req.Outpoint] = child + + // Mirror the historical record into r.pending so + // handleTerminated can carry over Trigger / ActorID + // without an extra store lookup, and so handleGetStatus + // answers from cache while the child runs. + r.pending[req.Outpoint] = cloneRegistryRecord(*existing) + + return fn.Ok[RegistryResp](&EnsureUnrollResp{ + ActorID: child.Ref().ID(), + Created: false, + }) + } + return fn.Ok[RegistryResp](&EnsureUnrollResp{ ActorID: existing.ActorID, Created: false, @@ -717,10 +824,19 @@ func stopChildAfterDrain(child *VTXOUnrollActor) { }() } -// restoreNonTerminal is the daemon's boot entry point for the unroll -// subsystem. It reads every record from the durable store that is not -// already Completed or Failed, spawns a fresh VTXOUnrollActor per -// target, and sends ResumeUnrollRequest to each. +// handleRestoreNonTerminal is the daemon's boot entry point for the +// unroll subsystem, dispatched through the registry actor's Receive loop +// so it shares the same goroutine as handleEnsure / handleGetStatus. +// +// Running inside the actor turn is what makes the r.active / r.pending +// mutations below race-free: NewUnrollRegistryActor calls Start() before +// the boot path issues the first restore, so by the time we get here the +// actor may already have processed concurrent Ensure / GetStatus +// messages from the chain resolver or RPC layer. +// +// It reads every record from the durable store that is not already +// Completed or Failed, spawns a fresh VTXOUnrollActor per target, and +// sends ResumeUnrollRequest to each. // // The per-target behavior then loads its checkpoint (proof, planner // state, sweep tx, last height), reconstructs the FSM in the same state @@ -731,23 +847,46 @@ func stopChildAfterDrain(child *VTXOUnrollActor) { // already-confirmed ones return immediately with their status. // // When restore fails for an individual target (spawn fails, or the -// resume Ask fails), we mark that target terminal with PhaseFailed and -// a descriptive reason rather than leaving the store entry non-terminal -// forever. A fresh Ensure from the chain resolver can then try again -// with a clean slate if the cause is transient. -func (r *registryBehavior) restoreNonTerminal(ctx context.Context) error { +// resume Ask fails), we leave the durable record non-terminal so that: +// +// - the next daemon restart will retry the restore from a clean +// slate when the transient cause is gone (e.g. a chain backend +// outage that prevented SubscribeBlocks / RegisterSpend on the +// previous boot), and +// +// - a fresh EnsureUnrollRequest for the same outpoint within the +// current boot will attempt an inline restore via handleEnsure +// (which detects "non-terminal record, no active child" and calls +// tryRestoreOne). +// +// Marking the record terminal on a transient restore failure would +// strand a recovery-critical job: ListNonTerminalRecords would skip it +// on every subsequent boot and handleEnsure would short-circuit on the +// terminal record. For a VTXO that is in unilateral_exit and near +// expiry, that translates into locked or lost funds — see issue #381. +func (r *registryBehavior) handleRestoreNonTerminal( + ctx context.Context) fn.Result[RegistryResp] { + records, err := r.cfg.Store.ListNonTerminalRecords(ctx) if err != nil { - return fmt.Errorf("list non-terminal records: %w", err) + return fn.Ok[RegistryResp](&restoreNonTerminalResp{ + Err: fmt. + Errorf("list non-terminal records: %w", err). + Error(), + }) } if len(records) == 0 { - return nil + return fn.Ok[RegistryResp](&restoreNonTerminalResp{}) } height, err := r.queryBestHeight(ctx) if err != nil { - return fmt.Errorf("best height for restore: %w", err) + return fn.Ok[RegistryResp](&restoreNonTerminalResp{ + Err: fmt. + Errorf("best height for restore: %w", err). + Error(), + }) } for i := range records { @@ -756,33 +895,59 @@ func (r *registryBehavior) restoreNonTerminal(ctx context.Context) error { continue } - child, err := r.spawn(ctx, record.TargetOutpoint) + child, err := r.tryRestoreOne(ctx, record, height) if err != nil { - _ = r.cfg.Store.MarkTerminal( - ctx, record.TargetOutpoint, PhaseFailed, - "spawn failed on restore: "+err.Error(), nil, + // Leave the record non-terminal so the next boot + // or the next EnsureUnrollRequest can retry. Log + // loudly: a persistent restore failure is a real + // problem even though it is recoverable. + r.log.WarnS(ctx, "Failed to restore unroll job; "+ + "record left non-terminal for retry", err, + slog.String( + "outpoint", + record.TargetOutpoint.String(), + ), + slog.String("actor_id", record.ActorID), ) continue } - _, err = child.Ref().Ask(ctx, &ResumeUnrollRequest{ - Height: height, - }).Await(ctx).Unpack() - if err != nil { - child.Stop() - _ = r.cfg.Store.MarkTerminal( - ctx, record.TargetOutpoint, PhaseFailed, - "resume failed on restore: "+err.Error(), nil, - ) + r.active[record.TargetOutpoint] = child - continue - } + // Mirror the historical record into r.pending so + // handleTerminated can carry over Trigger / ActorID + // without an extra store lookup, and so handleGetStatus + // answers from cache while the restored child runs. + r.pending[record.TargetOutpoint] = cloneRegistryRecord(record) + } - r.active[record.TargetOutpoint] = child + return fn.Ok[RegistryResp](&restoreNonTerminalResp{}) +} + +// tryRestoreOne spawns a fresh per-target actor for one non-terminal +// record and sends it a ResumeUnrollRequest. On any error the spawned +// child is stopped and the durable record is left untouched so the +// caller can retry (either via a future EnsureUnrollRequest or on the +// next daemon restart). +func (r *registryBehavior) tryRestoreOne(ctx context.Context, + record RegistryRecord, height int32) (*VTXOUnrollActor, error) { + + child, err := r.spawn(ctx, record.TargetOutpoint) + if err != nil { + return nil, fmt.Errorf("spawn failed on restore: %w", err) } - return nil + _, err = child.Ref().Ask(ctx, &ResumeUnrollRequest{ + Height: height, + }).Await(ctx).Unpack() + if err != nil { + child.Stop() + + return nil, fmt.Errorf("resume failed on restore: %w", err) + } + + return child, nil } // handlePersistActiveRecord is half of the two-message pair that drives diff --git a/unroll/registry_test.go b/unroll/registry_test.go index 6a1749899..8a5eaa9c5 100644 --- a/unroll/registry_test.go +++ b/unroll/registry_test.go @@ -1325,6 +1325,472 @@ func TestRegistryTerminalStatusRemainsQueryableWhilePersistBlocked( }, testTimeout, 10*time.Millisecond) } +// TestRegistryRestoreFailureLeavesRecordRetryable verifies that a +// transient failure during RestoreNonTerminal does NOT mark the durable +// record terminal: a subsequent RestoreNonTerminal call (e.g. on the +// next daemon boot, after the transient ChainSource / DB issue is +// resolved) must still find and resume the job. This is the regression +// guard for issue #381 ("Restore failure permanently disables unroll +// recovery"): an attacker or backend outage that fails the resume Ask +// on one boot must not strand a recovery-critical job forever. +func TestRegistryRestoreFailureLeavesRecordRetryable(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + store := newMemRegistryStore() + checkpoints := newMemCheckpointStore() + txconfirmRef := &fakeTxConfirmRef{} + + actorID := actorIDForTarget(proof.TargetOutpoint()) + err := store.UpsertRecord(t.Context(), RegistryRecord{ + TargetOutpoint: proof.TargetOutpoint(), + ActorID: actorID, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + require.NoError(t, err) + + registry := newRegistryHarnessWithSpawn(t, RegistryConfig{ + Store: store, + DeliveryStore: checkpoints, + ProofAssembler: &mockProofAssembler{proof: proof}, + VTXOStore: &mockVTXOStore{desc: desc}, + TxConfirmRef: txconfirmRef, + ChainSource: &fakeRegistryChainSourceRef{height: 201}, + Wallet: &fakeSweepWallet{}, + }) + t.Cleanup(registry.Stop) + + // First boot: install a spawnFunc that returns a child whose + // ResumeUnrollRequest fails (simulating a transient ChainSource + // outage on SubscribeBlocks / RegisterSpend during resume). + var attempts atomic.Int32 + failingSpawn := func(_ context.Context, target wire.OutPoint) ( + *VTXOUnrollActor, error) { + + attempts.Add(1) + behavior := actor.NewFunctionBehavior( + func(_ context.Context, msg Msg) fn.Result[Resp] { + if _, ok := msg.(*ResumeUnrollRequest); ok { + err := errors.New("transient chain " + + "outage") + + return fn.Err[Resp](err) + } + + return fn.Err[Resp]( + fmt.Errorf("unexpected msg %T", msg), + ) + }, + ) + + // Test children are owned by t.Cleanup after creation. + //nolint:contextcheck + return newTestUnrollChild(t, target, behavior), nil + } + registry.behavior.spawnFunc = failingSpawn + + // RestoreNonTerminal must NOT return an error and must NOT mark + // the record terminal; the durable record stays non-terminal so a + // future retry path can pick it up. + err = registry.RestoreNonTerminal(t.Context()) + require.NoError(t, err) + require.EqualValues(t, 1, attempts.Load()) + + record, err := store.GetRecord(t.Context(), proof.TargetOutpoint()) + require.NoError(t, err) + require.NotNil(t, record) + require.False( + t, record.IsTerminal(), + "restore failure must leave record non-terminal", + ) + require.Equal(t, PhaseMaterializing, record.Phase) + + // Second boot path: swap in a healthy spawnFunc that completes + // the ResumeUnrollRequest and verify RestoreNonTerminal now + // succeeds. The durable record must still be visible to + // ListNonTerminalRecords. + healthySpawn := func(_ context.Context, target wire.OutPoint) ( + *VTXOUnrollActor, error) { + + attempts.Add(1) + behavior := actor.NewFunctionBehavior( + func(_ context.Context, msg Msg) fn.Result[Resp] { + switch msg.(type) { + case *ResumeUnrollRequest: + return fn.Ok[Resp](&AckResp{}) + + case *GetStateRequest: + return fn.Ok[Resp](&GetStateResp{ + Started: true, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + + default: + return fn.Err[Resp]( + fmt.Errorf("unexpected msg %T", + msg), + ) + } + }, + ) + + // Test children are owned by t.Cleanup after creation. + //nolint:contextcheck + return newTestUnrollChild(t, target, behavior), nil + } + registry.behavior.spawnFunc = healthySpawn + + err = registry.RestoreNonTerminal(t.Context()) + require.NoError(t, err) + require.EqualValues( + t, 2, attempts.Load(), + "second RestoreNonTerminal must respawn the child", + ) + + // The job is now active and visible via GetStatus. + resp, err := registry.Ref().Ask(t.Context(), &GetStatusRequest{ + Outpoint: proof.TargetOutpoint(), + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + status, ok := resp.(*GetStatusResp) + require.True(t, ok) + require.True(t, status.Found) + require.True(t, status.Active) + require.Equal(t, PhaseMaterializing, status.Phase) +} + +// TestRegistryEnsureRestoresFailedNonTerminalRecord verifies that when a +// non-terminal record exists in the durable store but no child is +// active (because a prior RestoreNonTerminal hit a transient error and +// left the record retryable), a fresh EnsureUnrollRequest from the +// chain resolver or RPC layer kicks off an inline restore instead of +// silently returning Created=false with a dormant job. This is the +// second half of the issue #381 fix: handleEnsure used to short-circuit +// on any existing record without checking whether the in-memory child +// was actually live. +func TestRegistryEnsureRestoresFailedNonTerminalRecord(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + store := newMemRegistryStore() + checkpoints := newMemCheckpointStore() + txconfirmRef := &fakeTxConfirmRef{} + + actorID := actorIDForTarget(proof.TargetOutpoint()) + err := store.UpsertRecord(t.Context(), RegistryRecord{ + TargetOutpoint: proof.TargetOutpoint(), + ActorID: actorID, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + require.NoError(t, err) + + registry := newRegistryHarnessWithSpawn(t, RegistryConfig{ + Store: store, + DeliveryStore: checkpoints, + ProofAssembler: &mockProofAssembler{proof: proof}, + VTXOStore: &mockVTXOStore{desc: desc}, + TxConfirmRef: txconfirmRef, + ChainSource: &fakeRegistryChainSourceRef{height: 201}, + Wallet: &fakeSweepWallet{}, + }) + t.Cleanup(registry.Stop) + + // Force the prior RestoreNonTerminal to "fail" by simply skipping + // it: leave the store record in place but never wire up an active + // child. EnsureUnroll must detect the gap and restore inline. + var resumes atomic.Int32 + registry.behavior.spawnFunc = func(_ context.Context, + target wire.OutPoint) (*VTXOUnrollActor, error) { + + behavior := actor.NewFunctionBehavior( + func(_ context.Context, msg Msg) fn.Result[Resp] { + switch msg.(type) { + case *ResumeUnrollRequest: + resumes.Add(1) + + return fn.Ok[Resp](&AckResp{}) + + case *GetStateRequest: + return fn.Ok[Resp](&GetStateResp{ + Started: true, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + + default: + return fn.Err[Resp]( + fmt.Errorf("unexpected msg %T", + msg), + ) + } + }, + ) + + // Test children are owned by t.Cleanup after creation. + //nolint:contextcheck + return newTestUnrollChild(t, target, behavior), nil + } + + // Caller asks Ensure for the same outpoint. The pre-existing + // non-terminal record + no active child must trigger inline + // restore via ResumeUnrollRequest. Created=false because the + // job was admitted in a previous boot. + resp, err := registry.Ref().Ask(t.Context(), &EnsureUnrollRequest{ + Outpoint: proof.TargetOutpoint(), + Trigger: TriggerCriticalExpiry, + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + ensureResp, ok := resp.(*EnsureUnrollResp) + require.True(t, ok) + require.False( + t, ensureResp.Created, + "existing record must surface as Created=false", + ) + require.Equal(t, actorID, ensureResp.ActorID) + require.EqualValues( + t, 1, resumes.Load(), + "EnsureUnroll on a non-terminal record with no active "+ + "child must trigger inline resume", + ) + + // The job is now active. + statusResp, err := registry.Ref().Ask(t.Context(), &GetStatusRequest{ + Outpoint: proof.TargetOutpoint(), + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + status, ok := statusResp.(*GetStatusResp) + require.True(t, ok) + require.True(t, status.Active) +} + +// TestRegistryEnsureRetriesAfterInlineRestoreFailure verifies that an +// inline restore failure inside handleEnsure does not strand the job: +// a subsequent EnsureUnroll on the same outpoint must attempt restore +// again rather than short-circuiting on the dormant non-terminal +// record. Combined with the no-mark-terminal behavior in +// restoreNonTerminal, this means a transient backend outage is fully +// recoverable both on the next boot AND via a follow-up Ensure within +// the same boot. +func TestRegistryEnsureRetriesAfterInlineRestoreFailure(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + store := newMemRegistryStore() + checkpoints := newMemCheckpointStore() + txconfirmRef := &fakeTxConfirmRef{} + + actorID := actorIDForTarget(proof.TargetOutpoint()) + err := store.UpsertRecord(t.Context(), RegistryRecord{ + TargetOutpoint: proof.TargetOutpoint(), + ActorID: actorID, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + require.NoError(t, err) + + registry := newRegistryHarnessWithSpawn(t, RegistryConfig{ + Store: store, + DeliveryStore: checkpoints, + ProofAssembler: &mockProofAssembler{proof: proof}, + VTXOStore: &mockVTXOStore{desc: desc}, + TxConfirmRef: txconfirmRef, + ChainSource: &fakeRegistryChainSourceRef{height: 201}, + Wallet: &fakeSweepWallet{}, + }) + t.Cleanup(registry.Stop) + + // First Ensure: ResumeUnrollRequest fails (transient). + var attempts atomic.Int32 + registry.behavior.spawnFunc = func(_ context.Context, + target wire.OutPoint) (*VTXOUnrollActor, error) { + + attempts.Add(1) + behavior := actor.NewFunctionBehavior( + func(_ context.Context, msg Msg) fn.Result[Resp] { + if _, ok := msg.(*ResumeUnrollRequest); ok { + return fn.Err[Resp]( + errors.New( + "transient resume " + + "failure"), + ) + } + + return fn.Err[Resp]( + fmt.Errorf("unexpected msg %T", msg), + ) + }, + ) + + // Test children are owned by t.Cleanup after creation. + //nolint:contextcheck + return newTestUnrollChild(t, target, behavior), nil + } + + _, err = registry.Ref().Ask(t.Context(), &EnsureUnrollRequest{ + Outpoint: proof.TargetOutpoint(), + Trigger: TriggerCriticalExpiry, + }).Await(t.Context()).Unpack() + require.Error(t, err) + require.Contains(t, err.Error(), "restore existing record") + + // The durable record must NOT have been marked terminal by the + // failed inline restore. + record, err := store.GetRecord(t.Context(), proof.TargetOutpoint()) + require.NoError(t, err) + require.NotNil(t, record) + require.False(t, record.IsTerminal()) + + // Second Ensure with a healthy spawn must succeed. + registry.behavior.spawnFunc = func(_ context.Context, + target wire.OutPoint) (*VTXOUnrollActor, error) { + + attempts.Add(1) + behavior := actor.NewFunctionBehavior( + func(_ context.Context, msg Msg) fn.Result[Resp] { + switch msg.(type) { + case *ResumeUnrollRequest: + return fn.Ok[Resp](&AckResp{}) + + case *GetStateRequest: + return fn.Ok[Resp](&GetStateResp{ + Started: true, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + + default: + return fn.Err[Resp]( + fmt.Errorf("unexpected msg %T", + msg), + ) + } + }, + ) + + // Test children are owned by t.Cleanup after creation. + //nolint:contextcheck + return newTestUnrollChild(t, target, behavior), nil + } + + resp, err := registry.Ref().Ask(t.Context(), &EnsureUnrollRequest{ + Outpoint: proof.TargetOutpoint(), + Trigger: TriggerCriticalExpiry, + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + ensureResp, ok := resp.(*EnsureUnrollResp) + require.True(t, ok) + require.False(t, ensureResp.Created) + require.EqualValues(t, 2, attempts.Load()) +} + +// TestRegistryRestoreNonTerminalDispatchedThroughActor verifies that +// RestoreNonTerminal serializes its r.active / r.pending mutations with +// concurrent Ensure / GetStatus traffic by going through the registry +// actor's Receive loop. Running this test under -race is the actual +// guard: any direct mutation outside the actor goroutine while another +// Ensure is in flight would trip the detector. +func TestRegistryRestoreNonTerminalDispatchedThroughActor(t *testing.T) { + proof := buildLinearProof(t) + desc := testDescriptor(t, proof.TargetOutpoint(), proof.CSVDelay()) + store := newMemRegistryStore() + checkpoints := newMemCheckpointStore() + txconfirmRef := &fakeTxConfirmRef{} + + actorID := actorIDForTarget(proof.TargetOutpoint()) + err := store.UpsertRecord(t.Context(), RegistryRecord{ + TargetOutpoint: proof.TargetOutpoint(), + ActorID: actorID, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + require.NoError(t, err) + + registry := newRegistryHarnessWithSpawn(t, RegistryConfig{ + Store: store, + DeliveryStore: checkpoints, + ProofAssembler: &mockProofAssembler{proof: proof}, + VTXOStore: &mockVTXOStore{desc: desc}, + TxConfirmRef: txconfirmRef, + ChainSource: &fakeRegistryChainSourceRef{height: 201}, + Wallet: &fakeSweepWallet{}, + }) + t.Cleanup(registry.Stop) + + registry.behavior.spawnFunc = func(_ context.Context, + target wire.OutPoint) (*VTXOUnrollActor, error) { + + behavior := actor.NewFunctionBehavior( + func(_ context.Context, msg Msg) fn.Result[Resp] { + switch msg.(type) { + case *ResumeUnrollRequest: + return fn.Ok[Resp](&AckResp{}) + + case *StartUnrollRequest: + return fn.Ok[Resp](&AckResp{}) + + case *GetStateRequest: + return fn.Ok[Resp](&GetStateResp{ + Started: true, + Trigger: TriggerRestart, + Phase: PhaseMaterializing, + }) + + default: + return fn.Err[Resp]( + fmt.Errorf("unexpected msg %T", + msg), + ) + } + }, + ) + + // Test children are owned by t.Cleanup after creation. + //nolint:contextcheck + return newTestUnrollChild(t, target, behavior), nil + } + + // Drive RestoreNonTerminal and a concurrent GetStatus probe at the + // same time. Both end up in the actor mailbox; -race catches any + // behavior-side state still mutated outside the goroutine. + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + + require.NoError(t, registry.RestoreNonTerminal(t.Context())) + }() + + go func() { + defer wg.Done() + + // A GetStatus probe is a read-only message that lands on the + // same mailbox, so the registry serializes it against the + // restore turn. + _, _ = registry.Ref().Ask(t.Context(), &GetStatusRequest{ + Outpoint: proof.TargetOutpoint(), + }).Await(t.Context()).Unpack() + }() + + wg.Wait() + + // After restore, the record is active. + resp, err := registry.Ref().Ask(t.Context(), &GetStatusRequest{ + Outpoint: proof.TargetOutpoint(), + }).Await(t.Context()).Unpack() + require.NoError(t, err) + + status, ok := resp.(*GetStatusResp) + require.True(t, ok) + require.True(t, status.Found) + require.True(t, status.Active) + require.Equal(t, PhaseMaterializing, status.Phase) +} + var _ RegistryStore = (*memRegistryStore)(nil) var _ RegistryStore = (*flakyRegistryStore)(nil) var _ RegistryStore = (*terminalFlakyRegistryStore)(nil)