From 14cb083e974c85c9bcf97bc6ed947cd6ab42820b Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Wed, 1 Jul 2026 19:25:34 +0200 Subject: [PATCH 01/11] rpc/walletdkrpc: add ACTIVITY has_more and cursor pagination Add has_more + next_cursor to ActivityList and an opaque cursor to ListRequest so the ACTIVITY view can page by a stable keyset instead of a drifting offset. --- rpc/walletdkrpc/wallet.proto | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/rpc/walletdkrpc/wallet.proto b/rpc/walletdkrpc/wallet.proto index 1e332f2a6..793d95728 100644 --- a/rpc/walletdkrpc/wallet.proto +++ b/rpc/walletdkrpc/wallet.proto @@ -430,8 +430,17 @@ message ListRequest { // limit caps the response size. Zero means use the daemon default. uint32 limit = 4; - // offset is the pagination offset within the chosen view. + // offset is the pagination offset within the chosen view. It applies + // to the VTXOS and ONCHAIN views; the ACTIVITY view paginates by the + // opaque cursor below instead and ignores offset. uint32 offset = 5; + + // cursor is the opaque pagination token for the ACTIVITY view. Empty + // starts from the newest entry; otherwise it is the next_cursor + // returned by the previous ActivityList page. It is stable across + // concurrent inserts, so paging never skips or duplicates rows. + // Ignored for the VTXOS and ONCHAIN views. + string cursor = 6; } message ListResponse { @@ -456,9 +465,17 @@ message ActivityList { // entries are the unified, time-sorted wallet operations. repeated WalletEntry entries = 1; - // total is the total number of entries matching the filter before - // limit and offset are applied. + // total is the number of entries in this page. It is a page count, + // not a full-feed count: the feed is paged by an opaque cursor, so + // callers use has_more, not total, to decide whether to fetch again. uint32 total = 2; + + // has_more reports whether more entries exist after this page. + bool has_more = 3; + + // next_cursor is the opaque token to pass as ListRequest.cursor to + // fetch the next page. Empty when has_more is false. + string next_cursor = 4; } message VTXOInventory { From bfe08d5e1b7f6628eb35832dc74b6ef134aef3bd Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Wed, 1 Jul 2026 19:25:34 +0200 Subject: [PATCH 02/11] rpc/walletdkrpc: regenerate stubs for activity pagination Regenerated via make rpc. --- rpc/walletdkrpc/wallet.pb.go | 57 ++++++++++++++++++++++++++++++------ 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/rpc/walletdkrpc/wallet.pb.go b/rpc/walletdkrpc/wallet.pb.go index a1c6249f5..e417f8495 100644 --- a/rpc/walletdkrpc/wallet.pb.go +++ b/rpc/walletdkrpc/wallet.pb.go @@ -1607,8 +1607,16 @@ type ListRequest struct { Kinds []EntryKind `protobuf:"varint,3,rep,packed,name=kinds,proto3,enum=walletdkrpc.EntryKind" json:"kinds,omitempty"` // limit caps the response size. Zero means use the daemon default. Limit uint32 `protobuf:"varint,4,opt,name=limit,proto3" json:"limit,omitempty"` - // offset is the pagination offset within the chosen view. - Offset uint32 `protobuf:"varint,5,opt,name=offset,proto3" json:"offset,omitempty"` + // offset is the pagination offset within the chosen view. It applies + // to the VTXOS and ONCHAIN views; the ACTIVITY view paginates by the + // opaque cursor below instead and ignores offset. + Offset uint32 `protobuf:"varint,5,opt,name=offset,proto3" json:"offset,omitempty"` + // cursor is the opaque pagination token for the ACTIVITY view. Empty + // starts from the newest entry; otherwise it is the next_cursor + // returned by the previous ActivityList page. It is stable across + // concurrent inserts, so paging never skips or duplicates rows. + // Ignored for the VTXOS and ONCHAIN views. + Cursor string `protobuf:"bytes,6,opt,name=cursor,proto3" json:"cursor,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1678,6 +1686,13 @@ func (x *ListRequest) GetOffset() uint32 { return 0 } +func (x *ListRequest) GetCursor() string { + if x != nil { + return x.Cursor + } + return "" +} + type ListResponse struct { state protoimpl.MessageState `protogen:"open.v1"` // WIRE-BREAKING CHANGE: PR #440 had @@ -1791,9 +1806,15 @@ type ActivityList struct { state protoimpl.MessageState `protogen:"open.v1"` // entries are the unified, time-sorted wallet operations. Entries []*WalletEntry `protobuf:"bytes,1,rep,name=entries,proto3" json:"entries,omitempty"` - // total is the total number of entries matching the filter before - // limit and offset are applied. - Total uint32 `protobuf:"varint,2,opt,name=total,proto3" json:"total,omitempty"` + // total is the number of entries in this page. It is a page count, + // not a full-feed count: the feed is paged by an opaque cursor, so + // callers use has_more, not total, to decide whether to fetch again. + Total uint32 `protobuf:"varint,2,opt,name=total,proto3" json:"total,omitempty"` + // has_more reports whether more entries exist after this page. + HasMore bool `protobuf:"varint,3,opt,name=has_more,json=hasMore,proto3" json:"has_more,omitempty"` + // next_cursor is the opaque token to pass as ListRequest.cursor to + // fetch the next page. Empty when has_more is false. + NextCursor string `protobuf:"bytes,4,opt,name=next_cursor,json=nextCursor,proto3" json:"next_cursor,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1842,6 +1863,20 @@ func (x *ActivityList) GetTotal() uint32 { return 0 } +func (x *ActivityList) GetHasMore() bool { + if x != nil { + return x.HasMore + } + return false +} + +func (x *ActivityList) GetNextCursor() string { + if x != nil { + return x.NextCursor + } + return "" +} + type VTXOInventory struct { state protoimpl.MessageState `protogen:"open.v1"` // vtxos are the live spendable VTXOs in the wallet. Order is @@ -4664,21 +4699,25 @@ const file_wallet_proto_rawDesc = "" + "\foperation_id\x18\x01 \x01(\tR\voperationId\x12\x1d\n" + "\n" + "amount_sat\x18\x02 \x01(\x04R\tamountSat\x12!\n" + - "\fpayment_hash\x18\x03 \x01(\tR\vpaymentHash\"\xb7\x01\n" + + "\fpayment_hash\x18\x03 \x01(\tR\vpaymentHash\"\xcf\x01\n" + "\vListRequest\x12)\n" + "\x04view\x18\x01 \x01(\x0e2\x15.walletdkrpc.ListViewR\x04view\x12!\n" + "\fpending_only\x18\x02 \x01(\bR\vpendingOnly\x12,\n" + "\x05kinds\x18\x03 \x03(\x0e2\x16.walletdkrpc.EntryKindR\x05kinds\x12\x14\n" + "\x05limit\x18\x04 \x01(\rR\x05limit\x12\x16\n" + - "\x06offset\x18\x05 \x01(\rR\x06offset\"\xbc\x01\n" + + "\x06offset\x18\x05 \x01(\rR\x06offset\x12\x16\n" + + "\x06cursor\x18\x06 \x01(\tR\x06cursor\"\xbc\x01\n" + "\fListResponse\x127\n" + "\bactivity\x18\x01 \x01(\v2\x19.walletdkrpc.ActivityListH\x00R\bactivity\x122\n" + "\x05vtxos\x18\x02 \x01(\v2\x1a.walletdkrpc.VTXOInventoryH\x00R\x05vtxos\x127\n" + "\aonchain\x18\x03 \x01(\v2\x1b.walletdkrpc.OnchainHistoryH\x00R\aonchainB\x06\n" + - "\x04body\"X\n" + + "\x04body\"\x94\x01\n" + "\fActivityList\x122\n" + "\aentries\x18\x01 \x03(\v2\x18.walletdkrpc.WalletEntryR\aentries\x12\x14\n" + - "\x05total\x18\x02 \x01(\rR\x05total\"T\n" + + "\x05total\x18\x02 \x01(\rR\x05total\x12\x19\n" + + "\bhas_more\x18\x03 \x01(\bR\ahasMore\x12\x1f\n" + + "\vnext_cursor\x18\x04 \x01(\tR\n" + + "nextCursor\"T\n" + "\rVTXOInventory\x12-\n" + "\x05vtxos\x18\x01 \x03(\v2\x17.walletdkrpc.WalletVTXOR\x05vtxos\x12\x14\n" + "\x05total\x18\x02 \x01(\rR\x05total\"\xd4\x01\n" + From e75c4f93671f535f58b9de5e2dd20acd7a7cb2bc Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Wed, 1 Jul 2026 19:25:34 +0200 Subject: [PATCH 03/11] multi: read List(ACTIVITY) from the canonical activity store Cut the ACTIVITY read path over from the derive-on-read merge to the persisted activity store: List pages activity_entries by the immutable (created_at, canonical_id) keyset and returns has_more + next_cursor, so paging a concurrently-growing feed neither skips nor duplicates. The merge is retained as deriveActivity, used only by the startup backfill; a nil store falls back to it. Widen the daemon activity-store interface with ListEntries and add rowToWalletEntry, the lossless inverse of the projection mapping. SubscribeWallet's snapshot is store-backed for free. --- darepod/config.go | 22 ++- swapwallet/activity_dualwrite_test.go | 9 +- swapwallet/deps.go | 2 +- swapwallet/history.go | 162 ++++++++++++++++++- swapwallet/list_store_test.go | 219 ++++++++++++++++++++++++++ swapwallet/projector.go | 60 ++++++- swapwallet/projector_test.go | 10 ++ 7 files changed, 468 insertions(+), 16 deletions(-) create mode 100644 swapwallet/list_store_test.go diff --git a/darepod/config.go b/darepod/config.go index f95d0b4a3..7f4eb8ccc 100644 --- a/darepod/config.go +++ b/darepod/config.go @@ -16,6 +16,7 @@ import ( "github.com/lightninglabs/darepo-client/chainbackends" "github.com/lightninglabs/darepo-client/credit" "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/db/sqlc" mailboxpb "github.com/lightninglabs/darepo-client/mailbox/pb" "github.com/lightninglabs/darepo-client/metrics" "github.com/lightninglabs/darepo-client/oor" @@ -257,7 +258,7 @@ type Config struct { // top-level field (not under SwapWallet) because the subserver is // registered by build tag regardless of whether the operator supplied // a [swapwallet] config section. A nil value disables projection. - ActivityStore ActivityProjector `mapstructure:"-"` + ActivityStore ActivityStore `mapstructure:"-"` // MaxOperatorFeeSat caps the per-round operator fee the client // is willing to pay under the #270 seal-time fee handshake. @@ -643,15 +644,22 @@ type SwapBackend interface { ResumePending(ctx context.Context) } -// ActivityProjector persists wallet activity rows to the canonical activity -// log as their lifecycle advances. *db.ActivityPersistenceStore satisfies it; -// the walletdkrpc subserver's projector calls ProjectEntry from the existing -// emit sites and the startup backfill. The interface keeps the daemon-side -// store out of the swapwallet build-tag boundary and lets tests pass nil. -type ActivityProjector interface { +// ActivityStore is the walletdkrpc subserver's handle on the canonical +// activity log. *db.ActivityPersistenceStore satisfies it; the projector +// writes through ProjectEntry from the emit sites and the startup backfill, +// and the List read path pages current-state rows through ListEntries. The +// interface keeps the daemon-side store out of the swapwallet build-tag +// boundary and lets tests pass nil. +type ActivityStore interface { // ProjectEntry advances the activity row to the projected state and // records the transition, atomically. ProjectEntry(ctx context.Context, p db.ActivityProjection) error + + // ListEntries returns up to limit current-state rows newest-first, + // starting after the (cursorCreated, cursorID) keyset. A cursorCreated + // of 0 starts from the newest row. + ListEntries(ctx context.Context, cursorCreated int64, cursorID string, + limit int32) ([]sqlc.ActivityEntry, error) } // SwapWalletConfig configures the optional walletdkrpc subserver. The struct diff --git a/swapwallet/activity_dualwrite_test.go b/swapwallet/activity_dualwrite_test.go index 4af1bb3c6..05038f6dc 100644 --- a/swapwallet/activity_dualwrite_test.go +++ b/swapwallet/activity_dualwrite_test.go @@ -75,8 +75,13 @@ func TestBackfillMirrorsLegacyMerge(t *testing.T) { }, } - // Legacy merge result is the comparison oracle. - merged, err := h.listActivity(ctx, &walletdkrpc.ListRequest{Limit: 100}) + // The derive-on-read merge is the comparison oracle — it is what the + // backfill seeds the store from (listActivity now reads the store). + merged, err := h.deriveActivity( + ctx, &walletdkrpc.ListRequest{ + Limit: 100, + }, + ) require.NoError(t, err) wantStatus := make(map[string]int64) for _, e := range merged.GetEntries() { diff --git a/swapwallet/deps.go b/swapwallet/deps.go index 9eb99276e..cfb6e208b 100644 --- a/swapwallet/deps.go +++ b/swapwallet/deps.go @@ -195,7 +195,7 @@ type Deps struct { // writes each emitted WalletEntry through as state advances; the // startup backfill seeds it from the history collectors. Nil disables // projection. - ActivityStore darepod.ActivityProjector + ActivityStore darepod.ActivityStore } // resolveDeadline returns the effective wallet deadline, applying the diff --git a/swapwallet/history.go b/swapwallet/history.go index bb7ca840b..ceeb363e7 100644 --- a/swapwallet/history.go +++ b/swapwallet/history.go @@ -4,9 +4,12 @@ package swapwallet import ( "context" + "encoding/base64" "encoding/hex" + "errors" "fmt" "sort" + "strconv" "strings" "github.com/btcsuite/btcd/chaincfg/chainhash" @@ -93,14 +96,163 @@ func (h *history) List(ctx context.Context, req *walletdkrpc.ListRequest) ( } } -// listActivity returns the merged WalletEntry stream — the v1 unified -// history. The page size is capped at the daemon-level maximum so a -// malformed request cannot fan out unbounded work; sources are queried -// with the request's own limit so per-source pagination remains the -// per-source contract. +// errInvalidActivityCursor is returned when the ACTIVITY cursor token cannot be +// decoded into a keyset position. +var errInvalidActivityCursor = errors.New("invalid activity cursor") + +// listActivity returns an ACTIVITY page read from the canonical activity store, +// newest-first and paged by the opaque cursor. Because the store orders by the +// immutable (created_at_unix, canonical_id) keyset, paging is stable across +// concurrent inserts: it neither skips nor duplicates rows. When no store is +// wired (tests without a database) it falls back to the derive-on-read merge. func (h *history) listActivity(ctx context.Context, req *walletdkrpc.ListRequest) (*walletdkrpc.ActivityList, error) { + if h.deps.ActivityStore == nil { + return h.deriveActivity(ctx, req) + } + + limit := h.deps.resolveListLimit(req.GetLimit()) + kindFilter, err := buildKindFilter(req.GetKinds()) + if err != nil { + return nil, err + } + + cursorCreated, cursorID, err := decodeActivityCursor(req.GetCursor()) + if err != nil { + return nil, err + } + + pendingOnly := req.GetPendingOnly() + + // Scan the keyset, applying filters in Go, until limit+1 rows match so + // has_more can be computed with the standard extra-row trick even when + // filters skip store rows. The keyset advances by the last SCANNED row; + // next_cursor points at the last RETURNED row so the next page resumes + // exactly after it. + matched := make([]*walletdkrpc.WalletEntry, 0, limit+1) + lastCreated, lastID := cursorCreated, cursorID + for uint32(len(matched)) <= limit { + batch, err := h.deps.ActivityStore.ListEntries( + ctx, lastCreated, lastID, int32(limit)+1, + ) + if err != nil { + return nil, fmt.Errorf("list activity entries: %w", err) + } + if len(batch) == 0 { + break + } + + for _, row := range batch { + lastCreated, lastID = row.CreatedAtUnix, row.CanonicalID + + entry, err := rowToWalletEntry(row) + if err != nil { + return nil, fmt.Errorf("decode activity row "+ + "%q: %w", row.CanonicalID, err) + } + if !matchesActivityFilter( + entry, pendingOnly, kindFilter, + ) { + + continue + } + + matched = append(matched, entry) + if uint32(len(matched)) > limit { + break + } + } + + // A short page means the store has no more rows to scan. + if uint32(len(batch)) < uint32(limit)+1 { + break + } + } + + hasMore := uint32(len(matched)) > limit + if hasMore { + matched = matched[:limit] + } + + var nextCursor string + if hasMore { + last := matched[len(matched)-1] + nextCursor = encodeActivityCursor( + last.GetCreatedAtUnix(), last.GetId(), + ) + } + + return &walletdkrpc.ActivityList{ + Entries: matched, + Total: uint32(len(matched)), + HasMore: hasMore, + NextCursor: nextCursor, + }, nil +} + +// matchesActivityFilter reports whether an entry passes the pending_only and +// kind filters. It is the single-entry form of filterEntries, applied per row +// during the store keyset scan. +func matchesActivityFilter(e *walletdkrpc.WalletEntry, pendingOnly bool, + kindFilter map[walletdkrpc.EntryKind]struct{}) bool { + + if pendingOnly && + e.GetStatus() != walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING { + return false + } + if len(kindFilter) > 0 { + if _, ok := kindFilter[e.GetKind()]; !ok { + return false + } + } + + return true +} + +// encodeActivityCursor encodes the immutable keyset position +// (created_at_unix, canonical_id) as an opaque base64 token. +func encodeActivityCursor(created int64, id string) string { + raw := strconv.FormatInt(created, 10) + ":" + id + + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +// decodeActivityCursor decodes a cursor token back into its keyset position. An +// empty cursor decodes to the newest-first start position (0, ""). +func decodeActivityCursor(cursor string) (int64, string, error) { + if cursor == "" { + return 0, "", nil + } + + raw, err := base64.RawURLEncoding.DecodeString(cursor) + if err != nil { + return 0, "", fmt.Errorf("%w: %v", errInvalidActivityCursor, + err) + } + + created, id, ok := strings.Cut(string(raw), ":") + if !ok { + return 0, "", errInvalidActivityCursor + } + + createdUnix, err := strconv.ParseInt(created, 10, 64) + if err != nil { + return 0, "", fmt.Errorf("%w: %v", errInvalidActivityCursor, + err) + } + + return createdUnix, id, nil +} + +// deriveActivity returns the merged WalletEntry stream by re-joining the live +// sources on read. It is the pre-canonical-log path, retained only to seed the +// canonical store during the startup backfill; the RPC read path +// (listActivity) reads the store instead. The page size is capped at the +// daemon-level maximum so a malformed request cannot fan out unbounded work. +func (h *history) deriveActivity(ctx context.Context, + req *walletdkrpc.ListRequest) (*walletdkrpc.ActivityList, error) { + limit := h.deps.resolveListLimit(req.GetLimit()) kindFilter, err := buildKindFilter(req.GetKinds()) if err != nil { diff --git a/swapwallet/list_store_test.go b/swapwallet/list_store_test.go new file mode 100644 index 000000000..d1437e57c --- /dev/null +++ b/swapwallet/list_store_test.go @@ -0,0 +1,219 @@ +//go:build walletdkrpc && swapruntime + +package swapwallet + +import ( + "context" + "testing" + + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/rpc/walletdkrpc" + "github.com/lightningnetwork/lnd/clock" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// newStoreListFixture wires a history reader over a real in-memory activity +// store so the store-backed List read path can be exercised end to end. +func newStoreListFixture(t *testing.T) (*history, + *db.ActivityPersistenceStore) { + + t.Helper() + + testDB := db.NewTestDB(t) + store := db.NewStore( + testDB.DB, testDB.Queries, testDB.Backend(), btclog.Disabled, + ).NewActivityStore(clock.NewDefaultClock()) + + deps := &Deps{ActivityStore: store} + runtime := newRuntime(t.Context(), deps) + t.Cleanup(runtime.stop) + + return newHistory(deps, runtime), store +} + +// seedActivity projects one entry into the store via the production mapping. +func seedActivity(t *testing.T, store *db.ActivityPersistenceStore, id string, + kind walletdkrpc.EntryKind, status walletdkrpc.EntryStatus, + created int64) { + + t.Helper() + + proj, err := entryToProjection(&walletdkrpc.WalletEntry{ + Id: id, + Kind: kind, + Status: status, + AmountSat: 1000, + CreatedAtUnix: created, + UpdatedAtUnix: created, + }) + require.NoError(t, err) + require.NoError(t, store.ProjectEntry(context.Background(), proj)) +} + +func activityIDs(list *walletdkrpc.ActivityList) []string { + out := make([]string, 0, len(list.GetEntries())) + for _, e := range list.GetEntries() { + out = append(out, e.GetId()) + } + + return out +} + +// TestListActivityReadsStore verifies List pages the store newest-first and +// resumes via next_cursor with a correct has_more. +func TestListActivityReadsStore(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + seedActivity( + t, store, "a", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 100, + ) + seedActivity( + t, store, "b", walletdkrpc.EntryKind_ENTRY_KIND_RECV, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 200, + ) + seedActivity( + t, store, "c", walletdkrpc.EntryKind_ENTRY_KIND_DEPOSIT, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 300, + ) + + page1, err := h.listActivity(ctx, &walletdkrpc.ListRequest{Limit: 2}) + require.NoError(t, err) + require.Equal(t, []string{"c", "b"}, activityIDs(page1)) + require.True(t, page1.GetHasMore()) + require.NotEmpty(t, page1.GetNextCursor()) + + page2, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 2, + Cursor: page1.GetNextCursor(), + }) + require.NoError(t, err) + require.Equal(t, []string{"a"}, activityIDs(page2)) + require.False(t, page2.GetHasMore()) + require.Empty(t, page2.GetNextCursor()) +} + +// TestListActivityStablePaginationUnderInsert verifies the #781 acceptance +// criterion: a row inserted between page fetches never causes an existing row +// to be skipped or duplicated, because the cursor is an immutable keyset. +func TestListActivityStablePaginationUnderInsert(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + seedActivity( + t, store, "a", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 100, + ) + seedActivity( + t, store, "b", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 200, + ) + seedActivity( + t, store, "c", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 300, + ) + + page1, err := h.listActivity(ctx, &walletdkrpc.ListRequest{Limit: 2}) + require.NoError(t, err) + require.Equal(t, []string{"c", "b"}, activityIDs(page1)) + + // A new op lands between page fetches, newer than the page-1 cursor. + seedActivity( + t, store, "d", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 250, + ) + + page2, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 2, + Cursor: page1.GetNextCursor(), + }) + require.NoError(t, err) + + // Page 2 continues strictly older than the cursor: "a" is returned + // once, "b"/"c" are not duplicated, and the newer "d" is simply above + // this pagination pass (a fresh read would surface it at the top). + require.Equal(t, []string{"a"}, activityIDs(page2)) +} + +// TestListActivityFilters verifies pending_only and kind filters apply over the +// store keyset scan. +func TestListActivityFilters(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + seedActivity( + t, store, "send", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 100, + ) + seedActivity( + t, store, "recv", walletdkrpc.EntryKind_ENTRY_KIND_RECV, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 200, + ) + seedActivity( + t, store, "exit", walletdkrpc.EntryKind_ENTRY_KIND_EXIT, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 300, + ) + + pending, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 10, + PendingOnly: true, + }) + require.NoError(t, err) + require.Equal(t, []string{"exit", "send"}, activityIDs(pending)) + + recvOnly, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 10, + Kinds: []walletdkrpc.EntryKind{ + walletdkrpc.EntryKind_ENTRY_KIND_RECV, + }, + }) + require.NoError(t, err) + require.Equal(t, []string{"recv"}, activityIDs(recvOnly)) +} + +// TestListActivityRejectsBadCursor verifies a malformed cursor is a clean +// error. +func TestListActivityRejectsBadCursor(t *testing.T) { + t.Parallel() + + h, _ := newStoreListFixture(t) + + _, err := h.listActivity(context.Background(), &walletdkrpc.ListRequest{ + Cursor: "!!!not-base64!!!", + }) + require.ErrorIs(t, err, errInvalidActivityCursor) +} + +// TestRowToWalletEntryRoundTrip verifies a WalletEntry survives the +// project → store → row → rowToWalletEntry round trip unchanged. +func TestRowToWalletEntryRoundTrip(t *testing.T) { + t.Parallel() + + ctx := context.Background() + _, store := newStoreListFixture(t) + + entry := sampleWalletEntry() + + proj, err := entryToProjection(entry) + require.NoError(t, err) + require.NoError(t, store.ProjectEntry(ctx, proj)) + + row, err := store.GetEntry(ctx, entry.GetId()) + require.NoError(t, err) + + got, err := rowToWalletEntry(row) + require.NoError(t, err) + require.True( + t, proto.Equal(entry, got), + "reconstructed entry must equal the original", + ) +} diff --git a/swapwallet/projector.go b/swapwallet/projector.go index 66b3ef88f..2b65ebaf1 100644 --- a/swapwallet/projector.go +++ b/swapwallet/projector.go @@ -9,6 +9,7 @@ import ( "log/slog" "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/db/sqlc" "github.com/lightninglabs/darepo-client/rpc/walletdkrpc" "google.golang.org/protobuf/encoding/protojson" ) @@ -79,7 +80,7 @@ func (r *Runtime) backfillActivity(ctx context.Context) { projected int ) for { - list, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + list, err := h.deriveActivity(ctx, &walletdkrpc.ListRequest{ Limit: limit, Offset: offset, }) @@ -195,3 +196,60 @@ func hexBytesOrNil(s string) []byte { return b } + +// rowToWalletEntry reconstructs a WalletEntry from a stored current-state row — +// the inverse of entryToProjection, used by the store-backed List read path. +// It is lossless: every WalletEntry field has a backing column. BLOB handles +// are hex-encoded back, the confirmation height is widened, and the request +// oneof is decoded from its protojson form. +func rowToWalletEntry(row sqlc.ActivityEntry) (*walletdkrpc.WalletEntry, + error) { + + var request *walletdkrpc.WalletEntryRequest + if row.RequestJson != "" { + request = &walletdkrpc.WalletEntryRequest{} + if err := protojson.Unmarshal( + []byte(row.RequestJson), request, + ); err != nil { + return nil, fmt.Errorf("unmarshal request: %w", err) + } + } + + var confHeight int32 + if row.ConfirmationHeight.Valid { + confHeight = int32(row.ConfirmationHeight.Int64) + } + + entry := &walletdkrpc.WalletEntry{ + Id: row.CanonicalID, + Kind: walletdkrpc.EntryKind(row.Kind), + Status: walletdkrpc.EntryStatus(row.Status), + AmountSat: row.AmountSat, + FeeSat: row.FeeSat, + Counterparty: row.Counterparty, + Note: row.Note, + FailureReason: row.FailureReason, + Request: request, + Progress: &walletdkrpc.WalletEntryProgress{ + Phase: walletdkrpc.WalletEntryPhase( + row.Phase, + ), + PhaseLabel: row.PhaseLabel, + PaymentHash: hex.EncodeToString(row.PaymentHash), + Txid: hex.EncodeToString(row.Txid), + ConfirmationHeight: confHeight, + VtxoOutpoint: row.VtxoOutpoint, + }, + CreatedAtUnix: row.CreatedAtUnix, + UpdatedAtUnix: row.UpdatedAtUnix, + } + + // failure_code is presence-tracked on the wire: absent means "no + // failure", so only set it for a non-zero stored code. + if row.FailureCode != 0 { + code := walletdkrpc.EntryFailureCode(row.FailureCode) + entry.FailureCode = &code + } + + return entry, nil +} diff --git a/swapwallet/projector_test.go b/swapwallet/projector_test.go index 65aa5a847..b563addf4 100644 --- a/swapwallet/projector_test.go +++ b/swapwallet/projector_test.go @@ -9,6 +9,7 @@ import ( "time" "github.com/lightninglabs/darepo-client/db" + "github.com/lightninglabs/darepo-client/db/sqlc" "github.com/lightninglabs/darepo-client/rpc/walletdkrpc" "github.com/stretchr/testify/require" "google.golang.org/protobuf/encoding/protojson" @@ -54,6 +55,15 @@ func (f *fakeActivityProjector) count() int { return len(f.projected) } +// ListEntries satisfies darepod.ActivityStore. This fake exercises only the +// write path; the store-backed read path is tested against a real DB store, so +// this returns no rows. +func (f *fakeActivityProjector) ListEntries(_ context.Context, _ int64, + _ string, _ int32) ([]sqlc.ActivityEntry, error) { + + return nil, nil +} + // ids returns the set of canonical ids the fake has been asked to project. func (f *fakeActivityProjector) ids() map[string]bool { f.mu.Lock() From 04095a36eb31663f34242e61c724c12e28f8b524 Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 12:45:54 +0200 Subject: [PATCH 04/11] db: add CountActivityEntriesByStatus activity query The wallet Status summary's pending count must reflect the whole feed, but the paginated List path caps its total at one page. Add a COUNT-by-status query so the full pending count can be read directly. --- db/sqlc/queries/activity_log.sql | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/db/sqlc/queries/activity_log.sql b/db/sqlc/queries/activity_log.sql index 82baebae5..70231c501 100644 --- a/db/sqlc/queries/activity_log.sql +++ b/db/sqlc/queries/activity_log.sql @@ -56,6 +56,12 @@ INSERT INTO activity_events ( -- GetActivityEntry returns one entry by its canonical id. SELECT * FROM activity_entries WHERE canonical_id = $1; +-- name: CountActivityEntriesByStatus :one +-- CountActivityEntriesByStatus returns the number of current-state rows in the +-- given status. It backs the wallet status summary's pending count, which must +-- reflect the whole feed rather than a single paginated page. +SELECT COUNT(*) FROM activity_entries WHERE status = sqlc.arg(status); + -- name: ListActivityEntries :many -- ListActivityEntries returns entries newest-first, paged by the immutable -- (created_at_unix, canonical_id) cursor so a row that transitions in place From fab902c83bd3ab293b8a92345adcacc6f39d08fc Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 12:45:54 +0200 Subject: [PATCH 05/11] db: regenerate sqlc for the activity pending count Generated by make sqlc from the new CountActivityEntriesByStatus query. No schema change. --- db/sqlc/activity_log.sql.go | 14 ++++++++++++++ db/sqlc/querier.go | 4 ++++ 2 files changed, 18 insertions(+) diff --git a/db/sqlc/activity_log.sql.go b/db/sqlc/activity_log.sql.go index 0c0fc12cd..ba6a0b00b 100644 --- a/db/sqlc/activity_log.sql.go +++ b/db/sqlc/activity_log.sql.go @@ -39,6 +39,20 @@ func (q *Queries) AppendActivityEvent(ctx context.Context, arg AppendActivityEve return err } +const CountActivityEntriesByStatus = `-- name: CountActivityEntriesByStatus :one +SELECT COUNT(*) FROM activity_entries WHERE status = $1 +` + +// CountActivityEntriesByStatus returns the number of current-state rows in the +// given status. It backs the wallet status summary's pending count, which must +// reflect the whole feed rather than a single paginated page. +func (q *Queries) CountActivityEntriesByStatus(ctx context.Context, status int64) (int64, error) { + row := q.db.QueryRowContext(ctx, CountActivityEntriesByStatus, status) + var count int64 + err := row.Scan(&count) + return count, err +} + const GetActivityEntry = `-- name: GetActivityEntry :one SELECT canonical_id, kind, status, amount_sat, fee_sat, counterparty, note, phase, phase_label, failure_code, failure_reason, payment_hash, txid, confirmation_height, vtxo_outpoint, swap_session_id, ledger_txid, boarding_addr, request_json, created_at_unix, updated_at_unix FROM activity_entries WHERE canonical_id = $1 ` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 97c2b6a25..074862bda 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -16,6 +16,10 @@ type Querier interface { CancelVHTLCRecoveryJob(ctx context.Context, arg CancelVHTLCRecoveryJobParams) (int64, error) ClearPendingIntentAnchorByOutpoint(ctx context.Context, arg ClearPendingIntentAnchorByOutpointParams) error CompleteVHTLCRecoveryJob(ctx context.Context, arg CompleteVHTLCRecoveryJobParams) (int64, error) + // CountActivityEntriesByStatus returns the number of current-state rows in the + // given status. It backs the wallet status summary's pending count, which must + // reflect the whole feed rather than a single paginated page. + CountActivityEntriesByStatus(ctx context.Context, status int64) (int64, error) CountBoardingIntentsByStatus(ctx context.Context, status string) (int64, error) CountClientLedgerEntries(ctx context.Context) (int64, error) CountUnresolvedBoardingSweepInputs(ctx context.Context, txid []byte) (int64, error) From efacad5df35decab1958b3d29b1f56f0d57ab86e Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 12:45:54 +0200 Subject: [PATCH 06/11] multi: count pending activity entries off the full feed Status.pending_count read resp.GetActivity().GetTotal() from a single max-sized List page. After cursor pagination that total is a per-page count, so a wallet with more pending entries than one page under-reported its pending count. Add CountByStatus to the activity store and the darepod.ActivityStore interface, and route the count through history.countPending: it counts the store directly when wired and falls back to the derived merge total when it is not. --- darepod/config.go | 5 ++++ db/activity_store.go | 21 +++++++++++++++++ db/activity_store_test.go | 29 +++++++++++++++++++++++ swapwallet/history.go | 33 ++++++++++++++++++++++++++ swapwallet/list_store_test.go | 44 +++++++++++++++++++++++++++++++++++ swapwallet/projector_test.go | 8 +++++++ swapwallet/service.go | 13 ++++------- 7 files changed, 145 insertions(+), 8 deletions(-) diff --git a/darepod/config.go b/darepod/config.go index 7f4eb8ccc..22bc92bc2 100644 --- a/darepod/config.go +++ b/darepod/config.go @@ -660,6 +660,11 @@ type ActivityStore interface { // of 0 starts from the newest row. ListEntries(ctx context.Context, cursorCreated int64, cursorID string, limit int32) ([]sqlc.ActivityEntry, error) + + // CountByStatus returns the number of current-state rows in the given + // status. It backs the wallet status summary's full-feed pending count, + // which the paginated List path cannot report. + CountByStatus(ctx context.Context, status int64) (int64, error) } // SwapWalletConfig configures the optional walletdkrpc subserver. The struct diff --git a/db/activity_store.go b/db/activity_store.go index 1bdec905a..ea52edc1f 100644 --- a/db/activity_store.go +++ b/db/activity_store.go @@ -23,6 +23,9 @@ type ActivityStore interface { GetActivityEntry(ctx context.Context, canonicalID string) (sqlc.ActivityEntry, error) + CountActivityEntriesByStatus(ctx context.Context, + status int64) (int64, error) + ListActivityEntries(ctx context.Context, arg sqlc.ListActivityEntriesParams) ( []sqlc.ActivityEntry, @@ -196,6 +199,24 @@ func (s *ActivityPersistenceStore) GetEntry(ctx context.Context, return entry, err } +// CountByStatus returns the number of current-state rows in the given status. +// Unlike ListEntries it is not paginated, so it backs the wallet status +// summary's pending count with a true full-feed total. +func (s *ActivityPersistenceStore) CountByStatus(ctx context.Context, + status int64) (int64, error) { + + var count int64 + + err := s.db.ExecTx(ctx, ReadTxOption(), func(q ActivityStore) error { + var err error + count, err = q.CountActivityEntriesByStatus(ctx, status) + + return err + }) + + return count, err +} + // ListEntries returns up to limit current-state rows newest-first, starting // after the (cursorCreated, cursorID) keyset. A cursorCreated of 0 starts from // the newest row. The cursor is the immutable (created_at_unix, canonical_id) diff --git a/db/activity_store_test.go b/db/activity_store_test.go index 0ba772583..6349db1f9 100644 --- a/db/activity_store_test.go +++ b/db/activity_store_test.go @@ -76,6 +76,35 @@ func TestActivityStoreProjectInsertsEntryAndEvent(t *testing.T) { require.EqualValues(t, 1, events[0].Status) } +// TestActivityStoreCountByStatus verifies CountByStatus returns a full, +// unpaginated count of the rows in a given status — the primitive the wallet +// status summary's pending count relies on. +func TestActivityStoreCountByStatus(t *testing.T) { + t.Parallel() + + ctx := context.Background() + store := newActivityStoreForTest(t) + + // Two pending rows (status 1) and one complete row (status 2). + complete := sampleProjection("c1") + complete.Status = 2 + require.NoError(t, store.ProjectEntry(ctx, sampleProjection("p1"))) + require.NoError(t, store.ProjectEntry(ctx, sampleProjection("p2"))) + require.NoError(t, store.ProjectEntry(ctx, complete)) + + pending, err := store.CountByStatus(ctx, 1) + require.NoError(t, err) + require.EqualValues(t, 2, pending) + + completed, err := store.CountByStatus(ctx, 2) + require.NoError(t, err) + require.EqualValues(t, 1, completed) + + failed, err := store.CountByStatus(ctx, 3) + require.NoError(t, err) + require.EqualValues(t, 0, failed) +} + // TestActivityStoreProjectSuppressesUnchanged verifies that re-projecting an // identical state appends no new event, so the backfill and the swap monitor's // replay do not accumulate duplicate transitions in the append-only log. diff --git a/swapwallet/history.go b/swapwallet/history.go index ceeb363e7..35ce44e88 100644 --- a/swapwallet/history.go +++ b/swapwallet/history.go @@ -191,6 +191,39 @@ func (h *history) listActivity(ctx context.Context, }, nil } +// countPending returns the total number of in-flight (PENDING) activity +// entries. When the canonical store is wired it counts rows directly, so the +// result is a true full-feed count instead of the single-page total the +// paginated listActivity path reports. Without a store it derives the merged +// pending set and counts that, matching the deadline-overlay semantics of the +// derive path. +func (h *history) countPending(ctx context.Context) (uint32, error) { + if h.deps.ActivityStore == nil { + list, err := h.deriveActivity(ctx, &walletdkrpc.ListRequest{ + View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, + PendingOnly: true, + Limit: h.deps.resolveMaxListLimit(), + }) + if err != nil { + return 0, err + } + + return list.GetTotal(), nil + } + + count, err := h.deps.ActivityStore.CountByStatus( + ctx, int64(walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING), + ) + if err != nil { + return 0, err + } + if count < 0 { + count = 0 + } + + return uint32(count), nil +} + // matchesActivityFilter reports whether an entry passes the pending_only and // kind filters. It is the single-entry form of filterEntries, applied per row // during the store keyset scan. diff --git a/swapwallet/list_store_test.go b/swapwallet/list_store_test.go index d1437e57c..3c482f989 100644 --- a/swapwallet/list_store_test.go +++ b/swapwallet/list_store_test.go @@ -98,6 +98,50 @@ func TestListActivityReadsStore(t *testing.T) { require.Empty(t, page2.GetNextCursor()) } +// TestCountPendingReflectsFullFeed verifies countPending returns the full +// number of pending rows rather than the single-page total the paginated read +// path reports. This is the store-backed count behind the wallet status +// summary's pending count. +func TestCountPendingReflectsFullFeed(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + // Three pending rows plus one terminal row. + seedActivity( + t, store, "p1", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 100, + ) + seedActivity( + t, store, "p2", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 200, + ) + seedActivity( + t, store, "p3", walletdkrpc.EntryKind_ENTRY_KIND_RECV, + walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, 300, + ) + seedActivity( + t, store, "done", walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, 400, + ) + + // A single-page pending read caps its total at the page size, so it + // cannot stand in for the pending count. + page, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 1, + PendingOnly: true, + }) + require.NoError(t, err) + require.EqualValues(t, 1, page.GetTotal()) + require.True(t, page.GetHasMore()) + + // countPending reports every pending row regardless of page size. + count, err := h.countPending(ctx) + require.NoError(t, err) + require.EqualValues(t, 3, count) +} + // TestListActivityStablePaginationUnderInsert verifies the #781 acceptance // criterion: a row inserted between page fetches never causes an existing row // to be skipped or duplicated, because the cursor is an immutable keyset. diff --git a/swapwallet/projector_test.go b/swapwallet/projector_test.go index b563addf4..7f6107b18 100644 --- a/swapwallet/projector_test.go +++ b/swapwallet/projector_test.go @@ -64,6 +64,14 @@ func (f *fakeActivityProjector) ListEntries(_ context.Context, _ int64, return nil, nil } +// CountByStatus satisfies darepod.ActivityStore. The count path is tested +// against a real DB store, so this fake reports nothing. +func (f *fakeActivityProjector) CountByStatus(_ context.Context, _ int64) ( + int64, error) { + + return 0, nil +} + // ids returns the set of canonical ids the fake has been asked to project. func (f *fakeActivityProjector) ids() map[string]bool { f.mu.Lock() diff --git a/swapwallet/service.go b/swapwallet/service.go index 852476e32..8128484ec 100644 --- a/swapwallet/service.go +++ b/swapwallet/service.go @@ -418,17 +418,14 @@ func (s *Service) fetchBalance(ctx context.Context) ( return resp, nil } -// countPendingEntries asks the history merger for a pending-only page and -// returns the size as the wallet-level pending count. +// countPendingEntries returns the wallet-level count of in-flight entries. It +// delegates to the history merger's full-feed pending count rather than a +// List page total, which is capped at one page under cursor pagination. func (s *Service) countPendingEntries(ctx context.Context) (uint32, error) { - resp, err := s.history.List(ctx, &walletdkrpc.ListRequest{ - View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, - PendingOnly: true, - Limit: s.deps.resolveMaxListLimit(), - }) + count, err := s.history.countPending(ctx) if err != nil { return 0, fmt.Errorf("count pending: %w", err) } - return resp.GetActivity().GetTotal(), nil + return count, nil } From be609f4f942fc5b7989c0ab73e43707a5dc4f399 Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 12:45:54 +0200 Subject: [PATCH 07/11] multi: expose activity pagination on client surfaces The ACTIVITY view paginates by an opaque cursor, but the CLI, MCP, and SDK surfaces only carried offset and dropped has_more/next_cursor, so no client could fetch a second page. Thread cursor through the SDK ListRequest, the MCP activity tool, and the darepocli activity command, and surface has_more/next_cursor on the SDK ActivityList and the CLI table/expanded output. --- cmd/darepocli/darepoclicommands/cmd_list.go | 21 ++++++++++++++++-- cmd/darepocli/darepoclicommands/mcp_wallet.go | 13 ++++++----- .../darepoclicommands/mcp_wallet_test.go | 3 ++- .../darepoclicommands/wallet_table.go | 18 +++++++++++++++ sdk/walletdk/client.go | 1 + sdk/walletdk/convert.go | 6 +++-- sdk/walletdk/convert_test.go | 6 ++++- sdk/walletdk/types.go | 22 +++++++++++++++++-- 8 files changed, 77 insertions(+), 13 deletions(-) diff --git a/cmd/darepocli/darepoclicommands/cmd_list.go b/cmd/darepocli/darepoclicommands/cmd_list.go index ea8ee62ef..c4b274e4c 100644 --- a/cmd/darepocli/darepoclicommands/cmd_list.go +++ b/cmd/darepocli/darepoclicommands/cmd_list.go @@ -20,6 +20,7 @@ func newActivityCmd() *cobra.Command { " darepocli activity\n" + " darepocli activity --pending --kind send,recv\n" + " darepocli activity --format json\n" + + " darepocli activity --cursor \n" + " darepocli activity inspect ", Args: cobra.NoArgs, RunE: walletActivity, @@ -32,6 +33,8 @@ func newActivityCmd() *cobra.Command { cmd.Flags().Uint32("limit", 0, "page size; 0 uses the daemon default") cmd.Flags().Uint32("offset", 0, "pagination offset") + cmd.Flags().String("cursor", "", + "activity page token from a prior page's next_cursor") cmd.Flags().String("format", "table", "output format (table|expanded|x|json)") @@ -46,6 +49,7 @@ func walletActivity(cmd *cobra.Command, _ []string) error { kinds, _ := cmd.Flags().GetStringSlice("kind") limit, _ := cmd.Flags().GetUint32("limit") offset, _ := cmd.Flags().GetUint32("offset") + cursor, _ := cmd.Flags().GetString("cursor") format, _ := cmd.Flags().GetString("format") if err := validateListFormat( @@ -59,6 +63,7 @@ func walletActivity(cmd *cobra.Command, _ []string) error { PendingOnly: pending, Limit: limit, Offset: offset, + Cursor: cursor, } for _, k := range kinds { parsed, err := parseEntryKind(k) @@ -77,10 +82,22 @@ func walletActivity(cmd *cobra.Command, _ []string) error { switch format { case "", "table": - return printWalletActivityTable(resp) + if err := printWalletActivityTable( + resp, + ); err != nil { + return err + } + + return printWalletActivityNextPage(resp) case "expanded", "x": - return printWalletActivityExpanded(resp) + if err := printWalletActivityExpanded( + resp, + ); err != nil { + return err + } + + return printWalletActivityNextPage(resp) case "json": return printWalletProto(resp) diff --git a/cmd/darepocli/darepoclicommands/mcp_wallet.go b/cmd/darepocli/darepoclicommands/mcp_wallet.go index deafb458f..8f4ce4fd9 100644 --- a/cmd/darepocli/darepoclicommands/mcp_wallet.go +++ b/cmd/darepocli/darepoclicommands/mcp_wallet.go @@ -54,10 +54,11 @@ func registerMCPWalletQueryTools(s *mcp.Server, }) type activityArgs struct { - PendingOnly bool `json:"pending_only,omitempty" jsonschema:"filter to in-flight entries"` //nolint:ll - Kinds []string `json:"kinds,omitempty" jsonschema:"kind filter (send, recv, deposit, exit)"` //nolint:ll - Limit uint32 `json:"limit,omitempty" jsonschema:"page size; zero uses daemon default"` //nolint:ll - Offset uint32 `json:"offset,omitempty" jsonschema:"pagination offset"` //nolint:ll + PendingOnly bool `json:"pending_only,omitempty" jsonschema:"filter to in-flight entries"` //nolint:ll + Kinds []string `json:"kinds,omitempty" jsonschema:"kind filter (send, recv, deposit, exit)"` //nolint:ll + Limit uint32 `json:"limit,omitempty" jsonschema:"page size; zero uses daemon default"` //nolint:ll + Offset uint32 `json:"offset,omitempty" jsonschema:"pagination offset (vtxos/onchain views only)"` //nolint:ll + Cursor string `json:"cursor,omitempty" jsonschema:"activity page token; pass next_cursor to page"` //nolint:ll } mcp.AddTool(s, &mcp.Tool{ Name: "activity", @@ -67,6 +68,7 @@ func registerMCPWalletQueryTools(s *mcp.Server, req, err := buildWalletActivityRequest( args.PendingOnly, args.Kinds, args.Limit, args.Offset, + args.Cursor, ) if err != nil { return nil, nil, err @@ -268,13 +270,14 @@ func registerMCPWalletMutateTools(s *mcp.Server, // applying the same activity filter parsing the CLI uses so the two // surfaces stay in lockstep. func buildWalletActivityRequest(pendingOnly bool, kinds []string, limit, - offset uint32) (*walletdkrpc.ListRequest, error) { + offset uint32, cursor string) (*walletdkrpc.ListRequest, error) { req := &walletdkrpc.ListRequest{ View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, PendingOnly: pendingOnly, Limit: limit, Offset: offset, + Cursor: cursor, } for _, k := range kinds { diff --git a/cmd/darepocli/darepoclicommands/mcp_wallet_test.go b/cmd/darepocli/darepoclicommands/mcp_wallet_test.go index 82d31a526..d7c4b4036 100644 --- a/cmd/darepocli/darepoclicommands/mcp_wallet_test.go +++ b/cmd/darepocli/darepoclicommands/mcp_wallet_test.go @@ -125,7 +125,7 @@ func TestBuildWalletActivityRequestHappyPath(t *testing.T) { t.Parallel() req, err := buildWalletActivityRequest( - true, []string{"send", "recv"}, 50, 100, + true, []string{"send", "recv"}, 50, 100, "cursor-token", ) require.NoError(t, err) require.Equal( @@ -135,4 +135,5 @@ func TestBuildWalletActivityRequestHappyPath(t *testing.T) { require.Len(t, req.GetKinds(), 2) require.Equal(t, uint32(50), req.GetLimit()) require.Equal(t, uint32(100), req.GetOffset()) + require.Equal(t, "cursor-token", req.GetCursor()) } diff --git a/cmd/darepocli/darepoclicommands/wallet_table.go b/cmd/darepocli/darepoclicommands/wallet_table.go index 152940e4a..bd83fb2c5 100644 --- a/cmd/darepocli/darepoclicommands/wallet_table.go +++ b/cmd/darepocli/darepoclicommands/wallet_table.go @@ -43,6 +43,24 @@ func printWalletActivityExpanded(resp *walletdkrpc.ListResponse) error { return renderWalletActivityExpanded(os.Stdout, resp) } +// printWalletActivityNextPage writes a next-page hint to stdout when the +// activity feed has more entries. The human-facing table and expanded views +// omit the raw cursor otherwise, so without this line a caller has no way to +// discover the token needed to reach page two. +func printWalletActivityNextPage(resp *walletdkrpc.ListResponse) error { + activity := resp.GetActivity() + if !activity.GetHasMore() { + return nil + } + + _, err := fmt.Fprintf( + os.Stdout, "\nmore entries available; next page: --cursor %s\n", + activity.GetNextCursor(), + ) + + return err +} + // renderWalletActivityTable renders activity entries as a tabwriter table. func renderWalletActivityTable(out io.Writer, resp *walletdkrpc.ListResponse) error { diff --git a/sdk/walletdk/client.go b/sdk/walletdk/client.go index 267e8a716..261b6af45 100644 --- a/sdk/walletdk/client.go +++ b/sdk/walletdk/client.go @@ -419,6 +419,7 @@ func (c *Client) List(ctx context.Context, req ListRequest) (*ListResult, Kinds: kinds, Limit: req.Limit, Offset: req.Offset, + Cursor: req.Cursor, }) if err != nil { return nil, fmt.Errorf("list wallet entries: %w", err) diff --git a/sdk/walletdk/convert.go b/sdk/walletdk/convert.go index 3d78a8d4e..ded34b763 100644 --- a/sdk/walletdk/convert.go +++ b/sdk/walletdk/convert.go @@ -47,8 +47,10 @@ func listResultFromProto(view ListView, entries = append(entries, entryFromProto(e)) } out.Activity = &ActivityList{ - Entries: entries, - Total: activity.GetTotal(), + Entries: entries, + Total: activity.GetTotal(), + HasMore: activity.GetHasMore(), + NextCursor: activity.GetNextCursor(), } case ListViewVTXOs: diff --git a/sdk/walletdk/convert_test.go b/sdk/walletdk/convert_test.go index a3abea0a3..118576d6b 100644 --- a/sdk/walletdk/convert_test.go +++ b/sdk/walletdk/convert_test.go @@ -363,7 +363,9 @@ func TestListResultFromProtoActivity(t *testing.T) { Kind: recv, }, }, - Total: 42, + Total: 42, + HasMore: true, + NextCursor: "cursor-token", }, }, } @@ -373,6 +375,8 @@ func TestListResultFromProtoActivity(t *testing.T) { require.Nil(t, out.VTXOs) require.Nil(t, out.Onchain) require.Equal(t, uint32(42), out.Activity.Total) + require.True(t, out.Activity.HasMore) + require.Equal(t, "cursor-token", out.Activity.NextCursor) require.Len(t, out.Activity.Entries, 2) require.Equal(t, "hash1", out.Activity.Entries[0].ID) require.Equal(t, EntryKindSend, out.Activity.Entries[0].Kind) diff --git a/sdk/walletdk/types.go b/sdk/walletdk/types.go index a16b0455c..9a42c936e 100644 --- a/sdk/walletdk/types.go +++ b/sdk/walletdk/types.go @@ -286,8 +286,15 @@ type ListRequest struct { // Limit caps the page size; zero uses the daemon default. Limit uint32 - // Offset is the pagination offset within the chosen view. + // Offset is the pagination offset. It applies to the VTXOs and + // Onchain views; the Activity view paginates by Cursor and ignores + // Offset. Offset uint32 + + // Cursor is the opaque pagination token for the Activity view. Empty + // starts from the newest entry; otherwise pass the NextCursor returned + // by the previous ActivityList page. + Cursor string } // ListResult is a tagged union: exactly one of Activity, VTXOs, or @@ -312,7 +319,18 @@ type ListResult struct { // activity view. type ActivityList struct { Entries []Entry - Total uint32 + + // Total is the number of entries on this page, not a full-feed count: + // the feed is cursor-paged, so use HasMore to decide whether to fetch + // again. + Total uint32 + + // HasMore reports whether more entries exist after this page. + HasMore bool + + // NextCursor is the token to pass as ListRequest.Cursor to fetch the + // next page. Empty when HasMore is false. + NextCursor string } // VTXOInventory is the live VTXO inventory returned by the vtxos view. From 53c2bd284e5f16f51960cf0f4706de89295dff8a Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 17:29:02 +0200 Subject: [PATCH 08/11] swapwallet: harden the store-backed activity read path #842 cut List(ACTIVITY) over to the canonical store. These fixes make that read path correct, bounded, and forward-compatible: - Do not persist the synthetic boarding-unconfirmed row. It is ephemeral live state recomputed from GetBalance with no durable id, so a delete-free upsert store could never clear it once the deposit confirms under its real txid:vout id. It lingered as a phantom PENDING row in List and inflated the Status pending count. - Bound the per-request scan on a filtered page. pending_only/kinds are applied in Go after decode, so a selective filter over a large table could scan and protojson-decode the whole table for one page. Cap scanned rows per call and return a cursor to resume. - Reject a cursor whose decoded timestamp is <= 0 rather than let it collide with the return-all sentinel and silently restart paging. - Decode stored requests with DiscardUnknown so a row written by a newer daemon still decodes; a genuinely corrupt row still fails loudly rather than being silently skipped. --- swapwallet/history.go | 52 ++++++++++++++++++++++++++++++-- swapwallet/list_store_test.go | 56 +++++++++++++++++++++++++++++++++++ swapwallet/projector.go | 28 ++++++++++++++---- swapwallet/projector_test.go | 46 ++++++++++++++++++++++++++++ 4 files changed, 174 insertions(+), 8 deletions(-) diff --git a/swapwallet/history.go b/swapwallet/history.go index 35ce44e88..e10b2fa60 100644 --- a/swapwallet/history.go +++ b/swapwallet/history.go @@ -100,6 +100,22 @@ func (h *history) List(ctx context.Context, req *walletdkrpc.ListRequest) ( // decoded into a keyset position. var errInvalidActivityCursor = errors.New("invalid activity cursor") +// syntheticBoardingUnconfirmedID is the id of the derive-path-only row that +// represents an unconfirmed boarding deposit. It is recomputed live from +// GetBalance and has no durable identity, so it is deliberately kept out of the +// canonical store (see the projector's id guard): a delete-free upsert store +// could never clear it once the deposit confirms under its real txid:vout id. +const syntheticBoardingUnconfirmedID = "boarding-unconfirmed" + +// activityScanBudgetFactor bounds how many store rows a single filtered +// ACTIVITY page may scan (and protojson-decode), as a multiple of the page +// limit. Filters (pending_only/kinds) are applied in Go after decode, so a +// highly selective filter over a large table could otherwise scan the whole +// table for one page. When the budget is hit before the page fills, the call +// returns a short page plus a cursor so the caller resumes — bounding the work +// of any one request at the cost of an occasional extra round-trip. +const activityScanBudgetFactor = 8 + // listActivity returns an ACTIVITY page read from the canonical activity store, // newest-first and paged by the opaque cursor. Because the store orders by the // immutable (created_at_unix, canonical_id) keyset, paging is stable across @@ -132,6 +148,9 @@ func (h *history) listActivity(ctx context.Context, // exactly after it. matched := make([]*walletdkrpc.WalletEntry, 0, limit+1) lastCreated, lastID := cursorCreated, cursorID + scanBudget := int(limit) * activityScanBudgetFactor + scanned := 0 + budgetExhausted := false for uint32(len(matched)) <= limit { batch, err := h.deps.ActivityStore.ListEntries( ctx, lastCreated, lastID, int32(limit)+1, @@ -145,6 +164,7 @@ func (h *history) listActivity(ctx context.Context, for _, row := range batch { lastCreated, lastID = row.CreatedAtUnix, row.CanonicalID + scanned++ entry, err := rowToWalletEntry(row) if err != nil { @@ -168,6 +188,16 @@ func (h *history) listActivity(ctx context.Context, if uint32(len(batch)) < uint32(limit)+1 { break } + + // Stop once the scan budget is spent without filling the page. + // The last scanned row becomes the resume cursor so the caller + // continues rather than the server scanning the rest in one + // call. + if uint32(len(matched)) <= limit && scanned >= scanBudget { + budgetExhausted = true + + break + } } hasMore := uint32(len(matched)) > limit @@ -176,11 +206,18 @@ func (h *history) listActivity(ctx context.Context, } var nextCursor string - if hasMore { + switch { + case hasMore: last := matched[len(matched)-1] nextCursor = encodeActivityCursor( last.GetCreatedAtUnix(), last.GetId(), ) + + case budgetExhausted: + // The page did not fill but the store is not drained: resume + // strictly after the last scanned row. + hasMore = true + nextCursor = encodeActivityCursor(lastCreated, lastID) } return &walletdkrpc.ActivityList{ @@ -275,6 +312,17 @@ func decodeActivityCursor(cursor string) (int64, string, error) { err) } + // A real row always has a positive created_at_unix (ProjectEntry + // substitutes the clock when a projection omits it), and the + // empty-cursor "start from newest" sentinel is handled above. So a + // non-positive decoded timestamp is a forged or corrupt token: reject + // it loudly rather than let created_at_unix == 0 collide with the + // return-all sentinel in the keyset query and silently restart paging + // from the newest row. + if createdUnix <= 0 { + return 0, "", errInvalidActivityCursor + } + return createdUnix, id, nil } @@ -537,7 +585,7 @@ func (h *history) collectPendingBoardingEntries(ctx context.Context) ( return []*walletdkrpc.WalletEntry{ { - Id: "boarding-unconfirmed", + Id: syntheticBoardingUnconfirmedID, Kind: walletdkrpc.EntryKind_ENTRY_KIND_DEPOSIT, Status: walletdkrpc.EntryStatus_ENTRY_STATUS_PENDING, AmountSat: resp.GetBoardingUnconfirmedSat(), diff --git a/swapwallet/list_store_test.go b/swapwallet/list_store_test.go index 3c482f989..64363c389 100644 --- a/swapwallet/list_store_test.go +++ b/swapwallet/list_store_test.go @@ -4,6 +4,7 @@ package swapwallet import ( "context" + "fmt" "testing" "github.com/btcsuite/btclog/v2" @@ -237,6 +238,61 @@ func TestListActivityRejectsBadCursor(t *testing.T) { require.ErrorIs(t, err, errInvalidActivityCursor) } +// TestListActivityRejectsNonPositiveCursor verifies a cursor whose timestamp is +// zero or negative is rejected rather than silently colliding with the +// return-all sentinel and restarting paging from the newest row. +func TestListActivityRejectsNonPositiveCursor(t *testing.T) { + t.Parallel() + + h, _ := newStoreListFixture(t) + + for _, created := range []int64{0, -1} { + cursor := encodeActivityCursor(created, "x") + _, err := h.listActivity( + context.Background(), &walletdkrpc.ListRequest{ + Cursor: cursor, + }, + ) + require.ErrorIs(t, err, errInvalidActivityCursor) + } +} + +// TestListActivityBoundsFilteredScan verifies a selective filter over a large +// non-matching table does not scan the whole table in one request: the call +// stops at the scan budget and returns an empty page plus a cursor to resume, +// instead of decoding every row (the H-2 amplification cliff). +func TestListActivityBoundsFilteredScan(t *testing.T) { + t.Parallel() + + ctx := context.Background() + h, store := newStoreListFixture(t) + + // Seed far more terminal rows than one page's scan budget + // (limit * activityScanBudgetFactor), none of which match --pending. + const rows = 60 + for i := 0; i < rows; i++ { + seedActivity( + t, store, fmt.Sprintf("c%02d", i), + walletdkrpc.EntryKind_ENTRY_KIND_SEND, + walletdkrpc.EntryStatus_ENTRY_STATUS_COMPLETE, + int64(100+i), + ) + } + + page, err := h.listActivity(ctx, &walletdkrpc.ListRequest{ + Limit: 2, + PendingOnly: true, + }) + require.NoError(t, err) + + // The budget-bounded scan returns no matches but signals more work with + // a resume cursor, rather than draining the table (which would report + // has_more=false). + require.Empty(t, page.GetEntries()) + require.True(t, page.GetHasMore()) + require.NotEmpty(t, page.GetNextCursor()) +} + // TestRowToWalletEntryRoundTrip verifies a WalletEntry survives the // project → store → row → rowToWalletEntry round trip unchanged. func TestRowToWalletEntryRoundTrip(t *testing.T) { diff --git a/swapwallet/projector.go b/swapwallet/projector.go index 2b65ebaf1..7d7d856fb 100644 --- a/swapwallet/projector.go +++ b/swapwallet/projector.go @@ -27,8 +27,14 @@ func (r *Runtime) project(ctx context.Context, entry *walletdkrpc.WalletEntry) { } // A row with no canonical id cannot be keyed; skip it, matching the - // id-guard the pending tracker already applies. - if entry.GetId() == "" { + // id-guard the pending tracker already applies. The synthetic + // boarding-unconfirmed row is skipped for the same reason: it is + // ephemeral live state recomputed from GetBalance with no durable + // identity, so persisting it into a delete-free store would strand a + // PENDING row that never clears once the deposit confirms under its + // real txid:vout id. + if entry.GetId() == "" || + entry.GetId() == syntheticBoardingUnconfirmedID { return } @@ -199,16 +205,26 @@ func hexBytesOrNil(s string) []byte { // rowToWalletEntry reconstructs a WalletEntry from a stored current-state row — // the inverse of entryToProjection, used by the store-backed List read path. -// It is lossless: every WalletEntry field has a backing column. BLOB handles -// are hex-encoded back, the confirmation height is widened, and the request -// oneof is decoded from its protojson form. +// Every WalletEntry field has a backing column: BLOB handles are hex-encoded +// back, the confirmation height is widened, and the request oneof is decoded +// from its protojson form. The reconstruction is not byte-for-byte identical in +// one case: Progress is always materialized, so an entry projected with a nil +// Progress round-trips to a non-nil empty Progress (no current producer emits a +// nil-Progress row, so this is latent). +// +// request_json is decoded with DiscardUnknown so a row written by a newer +// daemon (carrying a WalletEntryRequest field this binary does not know) still +// decodes instead of failing the whole page. A genuinely malformed request +// still errors — a corrupt row is inconsistent state that should surface, not +// be silently skipped. func rowToWalletEntry(row sqlc.ActivityEntry) (*walletdkrpc.WalletEntry, error) { var request *walletdkrpc.WalletEntryRequest if row.RequestJson != "" { request = &walletdkrpc.WalletEntryRequest{} - if err := protojson.Unmarshal( + opts := protojson.UnmarshalOptions{DiscardUnknown: true} + if err := opts.Unmarshal( []byte(row.RequestJson), request, ); err != nil { return nil, fmt.Errorf("unmarshal request: %w", err) diff --git a/swapwallet/projector_test.go b/swapwallet/projector_test.go index 7f6107b18..d10037d22 100644 --- a/swapwallet/projector_test.go +++ b/swapwallet/projector_test.go @@ -268,3 +268,49 @@ func TestProjectAndEmitStoreErrorStillEmits(t *testing.T) { runtime.projectAndEmit(context.Background(), sampleWalletEntry()) require.Equal(t, "payment-hash", recvEntry(t, ch).GetId()) } + +// TestProjectAndEmitSkipsEphemeralBoardingRow verifies the synthetic +// boarding-unconfirmed row is emitted to subscribers but never persisted: it +// is ephemeral live state with no durable id, and a delete-free store could +// never clear it once the deposit confirms under its real txid:vout id. +func TestProjectAndEmitSkipsEphemeralBoardingRow(t *testing.T) { + t.Parallel() + + store := &fakeActivityProjector{} + runtime, ch := newProjectorRuntime(t, store) + + entry := sampleWalletEntry() + entry.Id = syntheticBoardingUnconfirmedID + runtime.projectAndEmit(context.Background(), entry) + + require.Equal( + t, syntheticBoardingUnconfirmedID, recvEntry(t, ch).GetId(), + ) + require.Equal(t, 0, store.count(), "ephemeral row must not be stored") +} + +// TestRowToWalletEntryDiscardsUnknownRequestFields verifies a stored request +// carrying a field this binary does not know (schema drift from a newer +// daemon) still decodes, while genuinely malformed JSON still fails loudly — +// a corrupt row is inconsistent state that must surface, not be skipped. +func TestRowToWalletEntryDiscardsUnknownRequestFields(t *testing.T) { + t.Parallel() + + forward := sqlc.ActivityEntry{ + CanonicalID: "a", + RequestJson: `{"lightningInvoice":{"invoice":"lnbc1"},` + + `"futureField":42}`, + } + got, err := rowToWalletEntry(forward) + require.NoError(t, err) + require.Equal( + t, "lnbc1", got.GetRequest().GetLightningInvoice().GetInvoice(), + ) + + corrupt := sqlc.ActivityEntry{ + CanonicalID: "b", + RequestJson: `{not valid json`, + } + _, err = rowToWalletEntry(corrupt) + require.Error(t, err, "a corrupt request row must fail loudly") +} From 4fea88d342ea44c370785f850d5a1b7ce1cd5ad7 Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 17:29:02 +0200 Subject: [PATCH 09/11] swapwallet: project forced unilateral-exit rows to the store forceUnroll emitted the EXIT row to live subscribers but never projected it, so a store-backed List missed a user-initiated unilateral exit until the next startup backfill. Route it through projectAndEmit off a cancel-safe context, mirroring the cooperative leave and credit-pay paths. The terminal transition remains backfill-only, consistent with the cooperative-leave handling. --- swapwallet/admin.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/swapwallet/admin.go b/swapwallet/admin.go index 0188431dd..d8b0b0483 100644 --- a/swapwallet/admin.go +++ b/swapwallet/admin.go @@ -219,9 +219,15 @@ func (s *Service) forceUnroll(ctx context.Context, req *walletdkrpc.ExitRequest, return nil, status.Errorf(status.Code(err), "exit: %v", err) } + // Project the new EXIT row into the canonical store as well as fanning + // it out, mirroring the cooperative-leave and credit-pay paths. Without + // this the store-backed List would miss a user-initiated unilateral + // exit until the next startup backfill. Use a cancel-safe context so a + // client disconnect after the accepted Unroll cannot drop the store + // write. entry := unilateralExitEntryStub(req.GetOutpoint()) s.runtime.trackPendingEntryWithoutTimeout(entry) - s.runtime.emit(entry) + s.runtime.projectAndEmit(context.WithoutCancel(ctx), entry) return &walletdkrpc.ExitResponse{ Created: resp.GetCreated(), From 9520b56833fe5c22dea0505bce4a5152f454714d Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 17:29:02 +0200 Subject: [PATCH 10/11] darepocli: drop the no-op activity --offset flag The ACTIVITY view paginates by cursor and ignores offset, but the CLI and MCP activity surfaces still advertised and forwarded --offset, so a caller passing it was silently served page one. Remove offset from the activity command and the MCP activity tool; cursor is the only activity pager. --- cmd/darepocli/darepoclicommands/cmd_list.go | 3 --- cmd/darepocli/darepoclicommands/mcp_wallet.go | 9 +++------ cmd/darepocli/darepoclicommands/mcp_wallet_test.go | 3 +-- 3 files changed, 4 insertions(+), 11 deletions(-) diff --git a/cmd/darepocli/darepoclicommands/cmd_list.go b/cmd/darepocli/darepoclicommands/cmd_list.go index c4b274e4c..29dcdfb23 100644 --- a/cmd/darepocli/darepoclicommands/cmd_list.go +++ b/cmd/darepocli/darepoclicommands/cmd_list.go @@ -32,7 +32,6 @@ func newActivityCmd() *cobra.Command { "filter by kind (send,recv,deposit,exit); repeatable") cmd.Flags().Uint32("limit", 0, "page size; 0 uses the daemon default") - cmd.Flags().Uint32("offset", 0, "pagination offset") cmd.Flags().String("cursor", "", "activity page token from a prior page's next_cursor") cmd.Flags().String("format", "table", @@ -48,7 +47,6 @@ func walletActivity(cmd *cobra.Command, _ []string) error { pending, _ := cmd.Flags().GetBool("pending") kinds, _ := cmd.Flags().GetStringSlice("kind") limit, _ := cmd.Flags().GetUint32("limit") - offset, _ := cmd.Flags().GetUint32("offset") cursor, _ := cmd.Flags().GetString("cursor") format, _ := cmd.Flags().GetString("format") @@ -62,7 +60,6 @@ func walletActivity(cmd *cobra.Command, _ []string) error { View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, PendingOnly: pending, Limit: limit, - Offset: offset, Cursor: cursor, } for _, k := range kinds { diff --git a/cmd/darepocli/darepoclicommands/mcp_wallet.go b/cmd/darepocli/darepoclicommands/mcp_wallet.go index 8f4ce4fd9..95c26a98d 100644 --- a/cmd/darepocli/darepoclicommands/mcp_wallet.go +++ b/cmd/darepocli/darepoclicommands/mcp_wallet.go @@ -57,7 +57,6 @@ func registerMCPWalletQueryTools(s *mcp.Server, PendingOnly bool `json:"pending_only,omitempty" jsonschema:"filter to in-flight entries"` //nolint:ll Kinds []string `json:"kinds,omitempty" jsonschema:"kind filter (send, recv, deposit, exit)"` //nolint:ll Limit uint32 `json:"limit,omitempty" jsonschema:"page size; zero uses daemon default"` //nolint:ll - Offset uint32 `json:"offset,omitempty" jsonschema:"pagination offset (vtxos/onchain views only)"` //nolint:ll Cursor string `json:"cursor,omitempty" jsonschema:"activity page token; pass next_cursor to page"` //nolint:ll } mcp.AddTool(s, &mcp.Tool{ @@ -67,8 +66,7 @@ func registerMCPWalletQueryTools(s *mcp.Server, args activityArgs) (*mcp.CallToolResult, any, error) { req, err := buildWalletActivityRequest( - args.PendingOnly, args.Kinds, args.Limit, args.Offset, - args.Cursor, + args.PendingOnly, args.Kinds, args.Limit, args.Cursor, ) if err != nil { return nil, nil, err @@ -269,14 +267,13 @@ func registerMCPWalletMutateTools(s *mcp.Server, // buildWalletActivityRequest translates MCP activityArgs into a ListRequest, // applying the same activity filter parsing the CLI uses so the two // surfaces stay in lockstep. -func buildWalletActivityRequest(pendingOnly bool, kinds []string, limit, - offset uint32, cursor string) (*walletdkrpc.ListRequest, error) { +func buildWalletActivityRequest(pendingOnly bool, kinds []string, limit uint32, + cursor string) (*walletdkrpc.ListRequest, error) { req := &walletdkrpc.ListRequest{ View: walletdkrpc.ListView_LIST_VIEW_ACTIVITY, PendingOnly: pendingOnly, Limit: limit, - Offset: offset, Cursor: cursor, } diff --git a/cmd/darepocli/darepoclicommands/mcp_wallet_test.go b/cmd/darepocli/darepoclicommands/mcp_wallet_test.go index d7c4b4036..e231cbe27 100644 --- a/cmd/darepocli/darepoclicommands/mcp_wallet_test.go +++ b/cmd/darepocli/darepoclicommands/mcp_wallet_test.go @@ -125,7 +125,7 @@ func TestBuildWalletActivityRequestHappyPath(t *testing.T) { t.Parallel() req, err := buildWalletActivityRequest( - true, []string{"send", "recv"}, 50, 100, "cursor-token", + true, []string{"send", "recv"}, 50, "cursor-token", ) require.NoError(t, err) require.Equal( @@ -134,6 +134,5 @@ func TestBuildWalletActivityRequestHappyPath(t *testing.T) { require.True(t, req.GetPendingOnly()) require.Len(t, req.GetKinds(), 2) require.Equal(t, uint32(50), req.GetLimit()) - require.Equal(t, uint32(100), req.GetOffset()) require.Equal(t, "cursor-token", req.GetCursor()) } From f1cb7b8df0bfbdd72c31b467f349614fe8ceb3ae Mon Sep 17 00:00:00 2001 From: Dario Anongba Varela Date: Thu, 2 Jul 2026 17:29:02 +0200 Subject: [PATCH 11/11] swapwallet: refresh the activity read-path scope docs The package doc still claimed the read path was unchanged and List still derived from the live merge. That is now false and masked the completeness gap. State that List and the SubscribeWallet snapshot read the store by keyset cursor, that deriveActivity is fallback and backfill only, and enumerate the producers still seeded only by the startup backfill. --- swapwallet/doc.go | 39 ++++++++++++++++++++++++++++----------- 1 file changed, 28 insertions(+), 11 deletions(-) diff --git a/swapwallet/doc.go b/swapwallet/doc.go index da68e78d4..8d13c4270 100644 --- a/swapwallet/doc.go +++ b/swapwallet/doc.go @@ -76,16 +76,33 @@ // injected via Deps.ActivityStore): activity_entries is the current-state // projection and activity_events is the append-only transition log. Writes // happen project-then-emit at the swap monitor loop, the cooperative-leave -// submit, and the deadline overlay, plus a one-time startup backfill from the -// collectors below. The read path is UNCHANGED for now: List and -// SubscribeWallet still derive from the live merge in history.go. A later -// change cuts List over to the store's keyset cursor and SubscribeWallet over -// to the activity_events event_seq cursor, then removes the merge. +// submit, the credit poll, the forced unilateral exit, and the deadline +// overlay, plus a one-time startup backfill from the collectors below. // -// Until that cutover, the canonical id stored for a cooperative-leave EXIT is -// still the consumed VTXO outpoint and a DEPOSIT is still keyed by txid:vout -// (or the synthetic boarding-unconfirmed row), so the same operation can hold -// different ids pending vs. confirmed — exactly the limitation above. Giving -// EXIT/DEPOSIT/on-chain-send a stable cross-lifecycle id needs the daemon-side -// hooks this V1 LIMITATIONS block describes and is tracked separately. +// The RPC read path now reads the store: List(ACTIVITY) pages activity_entries +// by the immutable (created_at_unix, canonical_id) keyset cursor, and +// SubscribeWallet's include_existing snapshot goes through the same +// store-backed List. deriveActivity (the live merge) is retained only for the +// store-less/test build and to seed the startup backfill. Because the store is +// ordered by the immutable created_at keyset, the feed is newest-by-creation, +// not newest-by-update. +// +// Consequences of the store-backed read that are tracked, not yet closed: +// - Producers without an ongoing projector — confirmed boarding DEPOSIT and +// daemon-side sweep/EXIT rows derived from ListTransactions — reach the +// store only via the startup backfill, so a newly-confirmed one appears in +// List after the next restart rather than immediately. +// - The synthetic boarding-unconfirmed DEPOSIT row is derive-path-only: it +// is ephemeral live state (recomputed from GetBalance, no durable id) and +// is deliberately NOT projected, so on a store build an unconfirmed +// boarding deposit surfaces via Balance rather than as an activity row +// until it confirms. +// +// The canonical id stored for a cooperative-leave EXIT is still the consumed +// VTXO outpoint and a DEPOSIT is still keyed by txid:vout, so the same +// operation can hold different ids pending vs. confirmed — exactly the +// limitation above. Giving EXIT/DEPOSIT/on-chain-send a stable cross-lifecycle +// id, and projecting the backfill-only producers on an ongoing basis, needs +// the daemon-side hooks this V1 LIMITATIONS block describes and is tracked +// separately. package swapwallet