swapwallet: project credit operations into the canonical activity log (#774)

swapwallet/credit_projector.go

@@ -127,7 +127,15 @@ func (r *Runtime) pollCreditOps(projected map[string]credit.State) {
 			r.trackPendingEntryWithoutTimeout(entry)
 		}
 
-		r.emit(entry)
+		// Project the credit row into the canonical activity log before
+		// fanning it out. Credit-only sends reach the feed only through
+		// this poll (never Runtime.emit from the swap monitor), so
+		// without this they would be absent from the store and vanish
+		// once the read path cuts over to it (issue #774; the #829
+		// class of bug). The store suppresses no-op re-projections, so
+		// the coarse re-poll of unchanged terminal rows appends no
+		// duplicate events.
+		r.projectAndEmit(r.rootCtx, entry)
 	}
 }
 

swapwallet/credit_projector_test.go

@@ -135,6 +135,52 @@ func TestCreditProjectorProjectsOwnedTerminals(t *testing.T) {
 	require.Empty(t, drainEntries(ch))
 }
 
+// TestCreditProjectorWritesToStore asserts the projector persists the credit
+// rows it owns into the canonical activity store (not only emits them), so
+// credit-only sends are in the store before the read path cuts over to it. A
+// re-poll of unchanged state projects nothing further.
+func TestCreditProjectorWritesToStore(t *testing.T) {
+	t.Parallel()
+
+	reg := &fakeCreditRegistry{
+		listResp: &credit.ListCreditOpsResponse{
+			Ops: []credit.CreditOpSummary{
+				{
+					OpID:       "op-pay",
+					OpKey:      "pay:" + payHashHex,
+					Kind:       credit.KindPay,
+					State:      credit.StateCompleted,
+					CreditOnly: true,
+					AmountSat:  500,
+				},
+				{
+					OpID:      "op-recv",
+					OpKey:     "recv:xyz",
+					Kind:      credit.KindReceive,
+					State:     credit.StateCompleted,
+					AmountSat: 42,
+				},
+			},
+		},
+	}
+	store := &fakeActivityProjector{}
+	deps := &Deps{CreditRegistry: reg, ActivityStore: store}
+	runtime := newRuntime(t.Context(), deps)
+	t.Cleanup(runtime.stop)
+
+	projected := make(map[string]credit.State)
+	runtime.pollCreditOps(projected)
+
+	require.Equal(t, 2, store.count())
+	ids := store.ids()
+	require.True(t, ids[payHashHex], "credit-only pay projected by hash")
+	require.True(t, ids["op-recv"], "credit receive projected by op id")
+
+	// A second poll with unchanged state projects nothing further.
+	runtime.pollCreditOps(projected)
+	require.Equal(t, 2, store.count())
+}
+
 // TestCreditProjectorProjectsFailure asserts a failed credit op surfaces as a
 // FAILED WalletEntry carrying the operation's terminal error.
 func TestCreditProjectorProjectsFailure(t *testing.T) {

swapwallet/projector_test.go

@@ -54,6 +54,19 @@ func (f *fakeActivityProjector) count() int {
 	return len(f.projected)
 }
 
+// ids returns the set of canonical ids the fake has been asked to project.
+func (f *fakeActivityProjector) ids() map[string]bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	out := make(map[string]bool, len(f.projected))
+	for _, p := range f.projected {
+		out[p.CanonicalID] = true
+	}
+
+	return out
+}
+
 // sampleWalletEntry builds a fully populated SEND WalletEntry fixture.
 func sampleWalletEntry() *walletdkrpc.WalletEntry {
 	return &walletdkrpc.WalletEntry{

swapwallet/router.go

@@ -297,10 +297,11 @@ func (r *router) sendCreditInvoiceIntent(ctx context.Context,
 	// Emit a pending entry keyed by the payment hash. A mixed pay's swap
 	// session updates this same row through the monitor loop; a credit-only
 	// pay completes server-side and the row is reconciled from credit
-	// state.
+	// state. Project off the RPC context so a CLI disconnect cannot cancel
+	// the write of an already-accepted pay.
 	entry := creditPayEntry(intent, paymentHash)
 	r.runtime.trackPendingEntryWithoutTimeout(entry)
-	r.runtime.emit(entry)
+	r.runtime.projectAndEmit(context.WithoutCancel(ctx), entry)
 
 	return &walletdkrpc.SendResponse{
 		Entry:           entry,