From dc383fd45c28c93ad9653d808e3e86ef548774e8 Mon Sep 17 00:00:00 2001 From: "Hayim.Shaul@ibm.com" Date: Mon, 17 Aug 2026 10:51:21 +0000 Subject: [PATCH] fix(tokens): publish token events only after commit Token add and delete events were published while the database transaction that produced them was still open, so a subscriber could observe a token that was later rolled back and never persisted. Buffer the events on DBTransaction and publish them from Commit, once the underlying commit succeeded, or from FlushEvents when the transaction is owned by the caller. Rollback discards them. Service.AppendValid applies a request to a transaction it does not own, so it now returns a PostCommit function that the owner of the transaction invokes once its commit succeeded. The finality listener calls it after committing and before waking the finality waiters, so a confirmed transaction is never observed with its token events still pending. Signed-off-by: Hayim.Shaul@ibm.com --- docs/services/tokens.md | 50 ++++- token/services/tokens/storage.go | 62 +++++- token/services/tokens/storage_test.go | 149 +++++++++++++++ token/services/tokens/tokens.go | 41 ++-- token/services/tokens/tokens_test.go | 177 ++++++++++++++++++ token/services/ttx/finality/listener.go | 18 +- token/services/ttx/finality/listener_test.go | 68 +++++++ .../ttx/finality/mock/tokens_service.go | 33 ++-- token/services/ttx/finality/recovery_test.go | 12 +- 9 files changed, 569 insertions(+), 41 deletions(-) diff --git a/docs/services/tokens.md b/docs/services/tokens.md index 1615b7ea30..1116787481 100644 --- a/docs/services/tokens.md +++ b/docs/services/tokens.md @@ -9,7 +9,7 @@ The internal [Service](../../token/services/tokens/tokens.go) is responsible for ### Core Responsibilities * **State Management**: Updating the local `TokenDB` to reflect the ledger's state (marking tokens as spendable, pending, or deleted). See [storage.go](../../token/services/tokens/storage.go). * **Typed Token Support**: Providing a unified way to handle multiple cryptographic token formats (e.g., cleartext vs. commitments) simultaneously via the [TypedToken](../../token/services/tokens/typed.go) structure. -* **Lifecycle Monitoring**: Notifying local listeners (via `events.Publisher`) when tokens are added or removed. +* **Lifecycle Monitoring**: Notifying local listeners (via `events.Publisher`) when tokens are added or removed, once the change has been committed. See [Event Publication and Transaction Boundaries](#event-publication-and-transaction-boundaries). * **Consistency**: Identifying and removing stale unspent tokens by cross-referencing local storage with the ledger via the [Network Service](../../token/services/network/network.go) (see `PruneInvalidUnspentTokens`). ### Marking Spent Tokens @@ -21,6 +21,54 @@ event is published for it. A failure reported by the underlying store is always regardless of whether the token was found locally, so that a transaction is never recorded as processed while its spends were not applied. +### Event Publication and Transaction Boundaries + +The `store-token` (`tokens.AddToken`) and `delete-token` (`tokens.DeleteToken`) events are +published **only after the database transaction that produced them has been committed**. A +subscriber therefore never observes a token that is later rolled back and never persisted. + +`DBTransaction.AppendToken` and `DBTransaction.DeleteToken` record their events on the +transaction — `DBTransaction.Notify` buffers, it does not publish — and: + +* `DBTransaction.Commit(ctx)` publishes them, and only if the commit succeeded. This covers + the transactions the service owns, obtained from `DBStorage.NewTransaction`. +* `DBTransaction.Rollback` discards them. +* `DBTransaction.FlushEvents(ctx)` publishes them explicitly. It is used when the transaction + is owned by the caller (`DBStorage.ContinueTransaction`), since the service is then not the + one that decides whether the transaction is committed. + +`Service.AppendValid` is the continued-transaction case: it applies a token request to a +transaction owned by the caller, and returns a `PostCommit` function alongside the error. The +caller must invoke it after its own commit succeeded, and must not invoke it when it rolls +back: + +```go +tx, err := ttxDB.NewTransaction() +if err != nil { + return errors.Wrapf(err, "failed creating new transaction [%s]", txID) +} +defer func() { + if tx != nil { + tx.Rollback() + } +}() + +publishTokenEvents, err := tokens.AppendValid(ctx, tx, token.RequestAnchor(txID), tr) +if err != nil { + return errors.Wrapf(err, "failed to append valid token request [%s]", txID) +} +if err := tx.Commit(); err != nil { + return errors.Wrapf(err, "failed commit [%s]", txID) +} +tx = nil + +// the tokens are durably stored, so the events may now be observed +publishTokenEvents(ctx) +``` + +`PostCommit` is never nil, so it can be called unconditionally on the success path, and +calling it more than once publishes nothing further. The in-repo caller is the finality +listener, see [finality/listener.go](../../token/services/ttx/finality/listener.go). ## Token Representations diff --git a/token/services/tokens/storage.go b/token/services/tokens/storage.go index 869e437d0b..fff0fc2236 100644 --- a/token/services/tokens/storage.go +++ b/token/services/tokens/storage.go @@ -116,6 +116,11 @@ type TokenToAppend struct { } // DBTransaction encapsulates a single atomic update to the token database. +// +// Events recorded while the transaction is open are buffered and published only +// once the transaction has been committed, so that a subscriber never observes a +// token that is later rolled back. A DBTransaction is not safe for concurrent +// use, like the underlying database transaction it wraps. type DBTransaction struct { // Notifier is used to publish events upon successful deletion or addition. Notifier events.Publisher @@ -123,6 +128,10 @@ type DBTransaction struct { Tx *tokendb.Transaction // TMSID is the TMS identifier for the transaction. TMSID token.TMSID + + // pending holds the events recorded so far, in the order they were recorded. + // They are published by FlushEvents and discarded by Rollback. + pending []*TokenProcessorEvent } // NewTransaction creates a new transaction wrapper. @@ -134,7 +143,9 @@ func NewTransaction(notifier events.Publisher, tx *tokendb.Transaction, tmsID to }, nil } -// DeleteToken removes a single token from the database and notifies listeners. +// DeleteToken removes a single token from the database and records a delete-token +// event for each of its owners. The events are published only after the transaction +// commits, see Notify and FlushEvents. // // Delete is idempotent: marking an unknown token as spent is not an error, so a // failure returned by Delete always signals a real storage failure and is @@ -176,7 +187,9 @@ func (t *DBTransaction) DeleteTokens(ctx context.Context, deletedBy string, ids return nil } -// AppendToken records a new token in the database and notifies listeners. +// AppendToken records a new token in the database and records an add-token event for +// each of its owners. The events are published only after the transaction commits, +// see Notify and FlushEvents. func (t *DBTransaction) AppendToken(ctx context.Context, tta TokenToAppend) error { q, err := token2.ToQuantity(tta.Tok.Quantity, tta.Precision) if err != nil { @@ -228,7 +241,11 @@ func (t *DBTransaction) AppendToken(ctx context.Context, tta TokenToAppend) erro return nil } -// Notify publishes a token-related event to the system's notification bus. +// Notify records a token-related event for publication on the system's notification +// bus. The event is not published here: it is buffered until the transaction that +// produced it has been committed, and then published by FlushEvents. Publishing +// inside the open transaction would let subscribers observe tokens that are never +// persisted. func (t *DBTransaction) Notify(ctx context.Context, topic string, tmsID token.TMSID, walletID string, tokenType token2.Type, txID string, index uint64) { if t.Notifier == nil { logger.WarnfContext(ctx, "cannot notify others!") @@ -244,18 +261,45 @@ func (t *DBTransaction) Notify(ctx context.Context, topic string, tmsID token.TM Index: index, }) - logger.DebugfContext(ctx, "publish new event %v", e) - t.Notifier.Publish(e) + logger.DebugfContext(ctx, "record new event %v", e) + t.pending = append(t.pending, e) +} + +// FlushEvents publishes the events recorded so far, in the order they were recorded, +// and empties the buffer. +// +// It must be called only after the transaction that produced the events has been +// successfully committed. Commit does this for transactions owned by this type; when +// the transaction is owned by the caller (see DBStorage.ContinueTransaction), the +// caller is responsible for calling FlushEvents after its own commit succeeds. +// Calling it more than once is safe: the buffer is empty after the first call. +func (t *DBTransaction) FlushEvents(ctx context.Context) { + pending := t.pending + t.pending = nil + for _, e := range pending { + logger.DebugfContext(ctx, "publish new event %v", e) + t.Notifier.Publish(e) + } } -// Rollback cancels all changes made in the transaction. +// Rollback cancels all changes made in the transaction and discards the events +// recorded for it, so that nothing is published for a transaction that never +// reached the store. func (t *DBTransaction) Rollback() error { + t.pending = nil + return t.Tx.Rollback() } -// Commit persists all changes made in the transaction. -func (t *DBTransaction) Commit() error { - return t.Tx.Commit() +// Commit persists all changes made in the transaction and, only if that succeeds, +// publishes the events recorded for it. +func (t *DBTransaction) Commit(ctx context.Context) error { + if err := t.Tx.Commit(); err != nil { + return err + } + t.FlushEvents(ctx) + + return nil } // SetSpendableFlag updates the spendable status for the given tokens in the database. diff --git a/token/services/tokens/storage_test.go b/token/services/tokens/storage_test.go index 6cc3bc1411..46e5e32c74 100644 --- a/token/services/tokens/storage_test.go +++ b/token/services/tokens/storage_test.go @@ -79,7 +79,12 @@ func TestTransaction_Notify(t *testing.T) { mockTx.GetTokenReturns(&token2.Token{Type: "TOK"}, []string{"alice"}, nil) err = tx.DeleteTokens(ctx, "me", ids) require.NoError(t, err) + // the transaction is still open, nothing may be published yet + assert.Equal(t, 0, pub.PublishCallCount()) + + require.NoError(t, tx.Commit(ctx)) assert.Equal(t, 1, pub.PublishCallCount()) + assert.Equal(t, tokens.DeleteToken, pub.PublishArgsForCall(0).Topic()) } func TestTransaction_AppendToken_Notify(t *testing.T) { @@ -101,6 +106,146 @@ func TestTransaction_AppendToken_Notify(t *testing.T) { } err = tx.AppendToken(ctx, tta) require.NoError(t, err) + // the transaction is still open, nothing may be published yet + assert.Equal(t, 0, pub.PublishCallCount()) + + require.NoError(t, tx.Commit(ctx)) + require.Equal(t, 1, pub.PublishCallCount()) + e := pub.PublishArgsForCall(0) + assert.Equal(t, tokens.AddToken, e.Topic()) + assert.Equal(t, tokens.TokenMessage{ + TMSID: tmsID, + WalletID: "wallet1", + TokenType: "TOK", + TxID: "tx1", + Index: 0, + }, e.Message()) +} + +// TestTransaction_AppendToken_NoEventBeforeCommit is the reproduction reported in +// issue #2183: an add-token event must not escape while the transaction that stored +// the token is still open, because the owner of that transaction may still roll it +// back — and a published event cannot be retracted. +func TestTransaction_AppendToken_NoEventBeforeCommit(t *testing.T) { + ctx := context.Background() + tmsID := token.TMSID{Network: "net", Channel: "ch", Namespace: "ns"} + mockTx := &mock.FakeTokenStoreTransaction{} + pub := &mock.FakePublisher{} + + tx, err := tokens.NewTransaction(pub, &tokendb.Transaction{TokenStoreTransaction: mockTx}, tmsID) + require.NoError(t, err) + + tta := tokens.TokenToAppend{ + TxID: "tx1", + Index: 0, + Tok: &token2.Token{Type: "TOK", Owner: []byte("alice"), Quantity: "0x64"}, + Precision: 64, + Owners: []string{"wallet1"}, + Flags: tokens.Flags{Mine: true}, + } + require.NoError(t, tx.AppendToken(ctx, tta)) + require.Equal(t, 0, pub.PublishCallCount()) + + // the owner of the transaction decides to roll back + require.NoError(t, tx.Rollback()) + assert.Equal(t, 1, mockTx.RollbackCallCount()) + assert.Equal(t, 0, pub.PublishCallCount()) +} + +// TestTransaction_Commit_PublishesRecordedEventsInOrder checks that every event +// recorded by the transaction is published on commit, once per owner, in the order +// in which it was recorded. +func TestTransaction_Commit_PublishesRecordedEventsInOrder(t *testing.T) { + ctx := context.Background() + tmsID := token.TMSID{Network: "net", Channel: "ch", Namespace: "ns"} + mockTx := &mock.FakeTokenStoreTransaction{} + pub := &mock.FakePublisher{} + + tx, err := tokens.NewTransaction(pub, &tokendb.Transaction{TokenStoreTransaction: mockTx}, tmsID) + require.NoError(t, err) + + require.NoError(t, tx.AppendToken(ctx, tokens.TokenToAppend{ + TxID: "tx1", + Index: 0, + Tok: &token2.Token{Type: "TOK", Owner: []byte("alice"), Quantity: "0x64"}, + Precision: 64, + Owners: []string{"wallet1", "wallet2"}, + Flags: tokens.Flags{Mine: true}, + })) + mockTx.GetTokenReturns(&token2.Token{Type: "TOK"}, []string{"wallet3"}, nil) + require.NoError(t, tx.DeleteTokens(ctx, "me", []*token2.ID{{TxId: "tx0", Index: 3}})) + require.Equal(t, 0, pub.PublishCallCount()) + + require.NoError(t, tx.Commit(ctx)) + require.Equal(t, 3, pub.PublishCallCount()) + + expected := []tokens.TokenMessage{ + {TMSID: tmsID, WalletID: "wallet1", TokenType: "TOK", TxID: "tx1", Index: 0}, + {TMSID: tmsID, WalletID: "wallet2", TokenType: "TOK", TxID: "tx1", Index: 0}, + {TMSID: tmsID, WalletID: "wallet3", TokenType: "TOK", TxID: "tx0", Index: 3}, + } + expectedTopics := []string{tokens.AddToken, tokens.AddToken, tokens.DeleteToken} + for i, msg := range expected { + e := pub.PublishArgsForCall(i) + assert.Equal(t, expectedTopics[i], e.Topic()) + assert.Equal(t, msg, e.Message()) + } +} + +// TestTransaction_Commit_Error_PublishesNothing checks that a failed commit publishes +// nothing: the tokens were not persisted, so no subscriber may learn about them. +func TestTransaction_Commit_Error_PublishesNothing(t *testing.T) { + ctx := context.Background() + tmsID := token.TMSID{Network: "net", Channel: "ch", Namespace: "ns"} + mockTx := &mock.FakeTokenStoreTransaction{} + pub := &mock.FakePublisher{} + + tx, err := tokens.NewTransaction(pub, &tokendb.Transaction{TokenStoreTransaction: mockTx}, tmsID) + require.NoError(t, err) + + require.NoError(t, tx.AppendToken(ctx, tokens.TokenToAppend{ + TxID: "tx1", + Index: 0, + Tok: &token2.Token{Type: "TOK", Owner: []byte("alice"), Quantity: "0x64"}, + Precision: 64, + Owners: []string{"wallet1"}, + Flags: tokens.Flags{Mine: true}, + })) + + mockTx.CommitReturns(assert.AnError) + require.ErrorIs(t, tx.Commit(ctx), assert.AnError) + assert.Equal(t, 0, pub.PublishCallCount()) +} + +// TestTransaction_FlushEvents_Idempotent checks that publishing the recorded events +// twice does not duplicate them. The owner of a continued transaction may hold on to +// the flush returned by AppendValid, so a second call must be harmless. +func TestTransaction_FlushEvents_Idempotent(t *testing.T) { + ctx := context.Background() + tmsID := token.TMSID{Network: "net", Channel: "ch", Namespace: "ns"} + mockTx := &mock.FakeTokenStoreTransaction{} + pub := &mock.FakePublisher{} + + tx, err := tokens.NewTransaction(pub, &tokendb.Transaction{TokenStoreTransaction: mockTx}, tmsID) + require.NoError(t, err) + + require.NoError(t, tx.AppendToken(ctx, tokens.TokenToAppend{ + TxID: "tx1", + Index: 0, + Tok: &token2.Token{Type: "TOK", Owner: []byte("alice"), Quantity: "0x64"}, + Precision: 64, + Owners: []string{"wallet1"}, + Flags: tokens.Flags{Mine: true}, + })) + + tx.FlushEvents(ctx) + require.Equal(t, 1, pub.PublishCallCount()) + + tx.FlushEvents(ctx) + assert.Equal(t, 1, pub.PublishCallCount()) + + // committing afterwards must not publish the events again either + require.NoError(t, tx.Commit(ctx)) assert.Equal(t, 1, pub.PublishCallCount()) } @@ -140,6 +285,10 @@ func TestTransaction_Notify_NoPublisher(t *testing.T) { tx, err := tokens.NewTransaction(nil, &tokendb.Transaction{TokenStoreTransaction: mockTx}, tmsID) require.NoError(t, err) tx.Notify(ctx, tokens.AddToken, tmsID, "wallet1", "TOK", "tx1", 0) + + // nothing was recorded, so the commit has nothing to publish either + require.NoError(t, tx.Commit(ctx)) + assert.Equal(t, 1, mockTx.CommitCallCount()) } func TestTransaction_Rollback(t *testing.T) { diff --git a/token/services/tokens/tokens.go b/token/services/tokens/tokens.go index b5b333aaa3..7d7f0734bb 100644 --- a/token/services/tokens/tokens.go +++ b/token/services/tokens/tokens.go @@ -91,19 +91,36 @@ func NewService(tmsID token.TMSID, TMSProvider TMSProvider, networkProvider Netw return &Service{tmsID: tmsID, TMSProvider: TMSProvider, NetworkProvider: networkProvider, Storage: storage, RequestsCache: requestsCache} } +// PostCommit publishes the token events recorded while the caller's transaction was open. +// +// The caller of AppendValid owns that transaction and therefore decides whether it is +// committed or rolled back. It must invoke the returned PostCommit once its commit has +// succeeded, and must not invoke it when the transaction is rolled back: only then do +// subscribers observe exactly the tokens that were persisted. It is never nil, so it can +// be called unconditionally on the success path. +type PostCommit = func(ctx context.Context) + +// noPostCommit is the PostCommit returned when there is nothing to publish. +func noPostCommit(context.Context) {} + // AppendValid extracts actions from a token request, applies them to the local storage, // and sets the transaction status to Confirmed. This is a convenience function that combines // Append with SetStatus for valid/confirmed transactions. -func (t *Service) AppendValid(ctx context.Context, tx dbdriver.Transaction, txID token.RequestAnchor, request *token.Request) (err error) { +// +// The passed transaction is owned by the caller and is neither committed nor finished here. +// The add-token and delete-token events produced by the applied actions are therefore not +// published either: they are buffered, and the returned PostCommit publishes them. The caller +// must invoke it after committing tx, see PostCommit. +func (t *Service) AppendValid(ctx context.Context, tx dbdriver.Transaction, txID token.RequestAnchor, request *token.Request) (postCommit PostCommit, err error) { if request == nil { logger.DebugfContext(ctx, "transaction [%s], no request found, skip it", txID) - return nil + return noPostCommit, nil } if request.Metadata == nil { logger.DebugfContext(ctx, "transaction [%s], no metadata found, skip it", txID) - return nil + return noPostCommit, nil } logger.DebugfContext(ctx, "check transaction exists") @@ -111,24 +128,24 @@ func (t *Service) AppendValid(ctx context.Context, tx dbdriver.Transaction, txID if err != nil { logger.ErrorfContext(ctx, "transaction [%s], failed to check existence in db [%s]", txID, err) - return errors.WithMessagef(err, "transaction [%s], failed to check existence in db", txID) + return noPostCommit, errors.WithMessagef(err, "transaction [%s], failed to check existence in db", txID) } if exists { logger.DebugfContext(ctx, "transaction [%s], exists in db, skipping", txID) - return nil + return noPostCommit, nil } toSpend, toAppend, err := t.getActions(ctx, txID, request) if err != nil { - return errors.WithMessagef(err, "transaction [%s], failed to extract actions", txID) + return noPostCommit, errors.WithMessagef(err, "transaction [%s], failed to extract actions", txID) } defer t.removeCachedTokenRequest(string(txID)) logger.DebugfContext(ctx, "transaction [%s] start db transaction", txID) ts, err := t.Storage.ContinueTransaction(tx) if err != nil { - return errors.WithMessagef(err, "transaction [%s], failed to start db transaction", txID) + return noPostCommit, errors.WithMessagef(err, "transaction [%s], failed to start db transaction", txID) } defer func() { if err == nil { @@ -145,19 +162,19 @@ func (t *Service) AppendValid(ctx context.Context, tx dbdriver.Transaction, txID for _, tta := range toAppend { err = ts.AppendToken(ctx, tta) if err != nil { - return errors.WithMessagef(err, "transaction [%s], failed to append token", txID) + return noPostCommit, errors.WithMessagef(err, "transaction [%s], failed to append token", txID) } } logger.DebugfContext(ctx, "delete spend tokens") err = ts.DeleteTokens(ctx, string(txID), toSpend) if err != nil { - return errors.WithMessagef(err, "transaction [%s], failed to delete tokens", txID) + return noPostCommit, errors.WithMessagef(err, "transaction [%s], failed to delete tokens", txID) } logger.DebugfContext(ctx, "ready to commit") - return nil + return ts.FlushEvents, nil } // CacheRequest extracts actions from a token request and caches them locally to avoid redundant parsing during the commit phase. @@ -225,7 +242,7 @@ func (t *Service) SetSpendableFlag(ctx context.Context, value bool, ids ...*toke return errors.Wrapf(err, "failed setting spendable flag") } - return tx.Commit() + return tx.Commit(ctx) } // SetSpendableBySupportedTokenTypes sets the spendable flag for all tokens that match the provided formats. @@ -241,7 +258,7 @@ func (t *Service) SetSpendableBySupportedTokenTypes(ctx context.Context, types [ return errors.WithMessagef(err, "error setting supported tokens") } - if err := tx.Commit(); err != nil { + if err := tx.Commit(ctx); err != nil { return errors.WithMessagef(err, "error committing transaction") } diff --git a/token/services/tokens/tokens_test.go b/token/services/tokens/tokens_test.go index 2e49a76499..7c29f023d8 100644 --- a/token/services/tokens/tokens_test.go +++ b/token/services/tokens/tokens_test.go @@ -12,6 +12,7 @@ import ( "github.com/LFDT-Panurus/panurus/token" "github.com/LFDT-Panurus/panurus/token/driver" + "github.com/LFDT-Panurus/panurus/token/services/storage/tokendb" "github.com/LFDT-Panurus/panurus/token/services/tokens" "github.com/LFDT-Panurus/panurus/token/services/tokens/mock" token2 "github.com/LFDT-Panurus/panurus/token/token" @@ -208,3 +209,179 @@ func TestParseRedeem(t *testing.T) { assert.True(t, store[0].Flags.Redeemed) assert.Empty(t, store[0].Owners) } + +// appendValidContext bundles the mocks needed to exercise AppendValid on a +// transaction owned by the caller, as the finality listener does. +type appendValidContext struct { + service *tokens.Service + store *mock.FakeTokenStore + tx *mock.FakeTokenStoreTransaction + pub *mock.FakePublisher + request *token.Request + tmsID token.TMSID +} + +func setupAppendValid(t *testing.T) *appendValidContext { + t.Helper() + + tmsID := token.TMSID{Network: "net", Channel: "ch", Namespace: "ns"} + store := &mock.FakeTokenStore{} + tx := &mock.FakeTokenStoreTransaction{} + pub := &mock.FakePublisher{} + + store.TransactionExistsReturns(false, nil) + store.ContinueTokenDBTransactionReturns(tx, nil) + // the token to spend is known locally and owned by wallet2 + tx.GetTokenReturns(&token2.Token{Type: "TOK"}, []string{"wallet2"}, nil) + + cache := &mock.FakeCache{} + cache.GetReturns(&tokens.CacheEntry{ + ToAppend: []tokens.TokenToAppend{{ + TxID: "tx1", + Index: 0, + Tok: &token2.Token{Type: "TOK", Owner: []byte("alice"), Quantity: "0x64"}, + Precision: 64, + Owners: []string{"wallet1"}, + Flags: tokens.Flags{Mine: true}, + }}, + ToSpend: []*token2.ID{{TxId: "tx0", Index: 1}}, + }, true) + + storage, err := tokens.NewDBStorage(pub, &tokendb.StoreService{TokenStore: store}, tmsID) + require.NoError(t, err) + + return &appendValidContext{ + service: tokens.NewService(tmsID, nil, nil, storage, cache), + store: store, + tx: tx, + pub: pub, + request: &token.Request{Anchor: "tx1", Metadata: &driver.TokenRequestMetadata{}}, + tmsID: tmsID, + } +} + +// TestAppendValid_PublishesOnlyOnPostCommit checks the contract that fixes issue #2183: +// AppendValid does not commit the caller's transaction, so it must not publish the token +// events either. They are published by the returned PostCommit, which the owner of the +// transaction invokes once its commit succeeded. +func TestAppendValid_PublishesOnlyOnPostCommit(t *testing.T) { + ctx := context.Background() + c := setupAppendValid(t) + + postCommit, err := c.service.AppendValid(ctx, nil, "tx1", c.request) + require.NoError(t, err) + require.NotNil(t, postCommit) + + // the tokens were stored and deleted in the caller's still-open transaction + require.Equal(t, 1, c.tx.StoreTokenCallCount()) + require.Equal(t, 1, c.tx.DeleteCallCount()) + // ... but nothing was published: the caller may still roll back + require.Equal(t, 0, c.pub.PublishCallCount()) + // AppendValid must not finish a transaction it does not own + assert.Equal(t, 0, c.tx.CommitCallCount()) + + postCommit(ctx) + require.Equal(t, 2, c.pub.PublishCallCount()) + assert.Equal(t, tokens.AddToken, c.pub.PublishArgsForCall(0).Topic()) + assert.Equal(t, tokens.TokenMessage{ + TMSID: c.tmsID, + WalletID: "wallet1", + TokenType: "TOK", + TxID: "tx1", + Index: 0, + }, c.pub.PublishArgsForCall(0).Message()) + assert.Equal(t, tokens.DeleteToken, c.pub.PublishArgsForCall(1).Topic()) + assert.Equal(t, tokens.TokenMessage{ + TMSID: c.tmsID, + WalletID: "wallet2", + TokenType: "TOK", + TxID: "tx0", + Index: 1, + }, c.pub.PublishArgsForCall(1).Message()) +} + +// TestAppendValid_NeverPublishingPostCommit checks that every path that applies nothing +// still returns a usable PostCommit, so that callers can invoke it unconditionally on +// their success path, and that invoking it publishes nothing. +func TestAppendValid_NeverPublishingPostCommit(t *testing.T) { + ctx := context.Background() + + tests := []struct { + name string + setup func(c *appendValidContext) *token.Request + expectedErr string + }{ + { + name: "no request", + setup: func(c *appendValidContext) *token.Request { return nil }, + }, + { + name: "no metadata", + setup: func(c *appendValidContext) *token.Request { + return &token.Request{Anchor: "tx1"} + }, + }, + { + name: "transaction already applied", + setup: func(c *appendValidContext) *token.Request { + c.store.TransactionExistsReturns(true, nil) + + return c.request + }, + }, + { + name: "existence check fails", + setup: func(c *appendValidContext) *token.Request { + c.store.TransactionExistsReturns(false, assert.AnError) + + return c.request + }, + expectedErr: "failed to check existence in db", + }, + { + name: "continuing the transaction fails", + setup: func(c *appendValidContext) *token.Request { + c.store.ContinueTokenDBTransactionReturns(nil, assert.AnError) + + return c.request + }, + expectedErr: "failed to start db transaction", + }, + { + name: "appending a token fails", + setup: func(c *appendValidContext) *token.Request { + c.tx.StoreTokenReturns(assert.AnError) + + return c.request + }, + expectedErr: "failed to append token", + }, + { + name: "deleting a spent token fails", + setup: func(c *appendValidContext) *token.Request { + c.tx.DeleteReturns(assert.AnError) + + return c.request + }, + expectedErr: "failed to delete tokens", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + c := setupAppendValid(t) + request := test.setup(c) + + postCommit, err := c.service.AppendValid(ctx, nil, "tx1", request) + if test.expectedErr != "" { + require.ErrorContains(t, err, test.expectedErr) + } else { + require.NoError(t, err) + } + + require.NotNil(t, postCommit) + postCommit(ctx) + assert.Equal(t, 0, c.pub.PublishCallCount()) + }) + } +} diff --git a/token/services/ttx/finality/listener.go b/token/services/ttx/finality/listener.go index 22f4181da6..2702d59dbf 100644 --- a/token/services/ttx/finality/listener.go +++ b/token/services/ttx/finality/listener.go @@ -46,7 +46,10 @@ type tokenRequestHasher interface { //go:generate counterfeiter -o mock/tokens_service.go -fake-name TokensService . tokensService type tokensService interface { GetCachedTokenRequest(txID string) (*token.Request, []byte) - AppendValid(ctx context.Context, tx dbdriver.Transaction, anchor token.RequestAnchor, tr *token.Request) error + // AppendValid applies the token request to the passed transaction, which it does not + // commit. The returned function publishes the token events produced by the request and + // must be invoked by the owner of the transaction, and only once the commit succeeded. + AppendValid(ctx context.Context, tx dbdriver.Transaction, anchor token.RequestAnchor, tr *token.Request) (func(ctx context.Context), error) } type Listener struct { @@ -206,7 +209,8 @@ func Commit( } }() - if err := tokens.AppendValid(ctx, tx, token.RequestAnchor(txID), tr); err != nil { + publishTokenEvents, err := tokens.AppendValid(ctx, tx, token.RequestAnchor(txID), tr) + if err != nil { logger.ErrorfContext(ctx, "failed to append valid token request to token db [%s]: [%s]", txID, err) return errors.Wrapf(err, "failed to append valid token request to token db [%s]", txID) @@ -225,6 +229,16 @@ func Commit( tx = nil + // The tokens are durably stored only now, so this is the first point at which + // the add-token and delete-token events they produced may be observed. Publish + // them before the status event, so that a woken finality waiter does not see a + // confirmed transaction whose token events are still pending. The nil check + // guards against an implementation that returns none: a missing publication is + // preferable to panicking in the listener's goroutine. + if publishTokenEvents != nil { + publishTokenEvents(ctx) + } + // The transactional SetStatus above bypasses the store service, so push // the status event explicitly — otherwise finality waiters only wake on // the fallback polling. diff --git a/token/services/ttx/finality/listener_test.go b/token/services/ttx/finality/listener_test.go index 78c61cdbfb..d0f35f041d 100644 --- a/token/services/ttx/finality/listener_test.go +++ b/token/services/ttx/finality/listener_test.go @@ -249,6 +249,74 @@ func TestCommit_NoNotifyOnCommitFailure(t *testing.T) { assert.Equal(t, 0, db.NotifyStatusCallCount(), "a failed commit must not wake waiters") } +// TestCommit_PublishesTokenEventsAfterCommit verifies that the token events produced by +// AppendValid are published by Commit, and only once the store transaction is committed: +// before that, the tokens they refer to may still be rolled back. +func TestCommit_PublishesTokenEventsAfterCommit(t *testing.T) { + db := &mock.TransactionDB{} + storeTx := &drivermock.TransactionStoreTransaction{} + db.NewTransactionReturns(storeTx, nil) + + var publishedAfter []string + tokens := &mock.TokensService{} + tokens.AppendValidReturns(func(context.Context) { + publishedAfter = append(publishedAfter, "publish") + }, nil) + db.NotifyStatusCalls(func(context.Context, string, storage.TxStatus, string) { + publishedAfter = append(publishedAfter, "notify-status") + }) + storeTx.CommitCalls(func() error { + publishedAfter = append(publishedAfter, "commit") + + return nil + }) + + require.NoError(t, finality.Commit(t.Context(), logging.MustGetLogger(), tokens, db, "tx1", nil)) + + // the token events go out after the commit, and before waiters are woken + assert.Equal(t, []string{"commit", "publish", "notify-status"}, publishedAfter) +} + +// TestCommit_NoTokenEventsWhenTransactionIsNotCommitted verifies that a transaction that +// never reaches the store publishes no token events: a subscriber must not observe tokens +// that were rolled back. This is the regression test for issue #2183 at the caller level. +func TestCommit_NoTokenEventsWhenTransactionIsNotCommitted(t *testing.T) { + tests := []struct { + name string + setup func(storeTx *drivermock.TransactionStoreTransaction) + }{ + { + name: "commit fails", + setup: func(storeTx *drivermock.TransactionStoreTransaction) { + storeTx.CommitReturns(errors.New("commit failed")) + }, + }, + { + name: "setting the status fails", + setup: func(storeTx *drivermock.TransactionStoreTransaction) { + storeTx.SetStatusReturns(errors.New("set status failed")) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + db := &mock.TransactionDB{} + storeTx := &drivermock.TransactionStoreTransaction{} + db.NewTransactionReturns(storeTx, nil) + test.setup(storeTx) + + published := 0 + tokens := &mock.TokensService{} + tokens.AppendValidReturns(func(context.Context) { published++ }, nil) + + require.Error(t, finality.Commit(t.Context(), logging.MustGetLogger(), tokens, db, "tx1", nil)) + assert.Equal(t, 0, published, "a transaction that was not committed must publish nothing") + assert.Equal(t, 1, storeTx.RollbackCallCount()) + }) + } +} + // TestOnError tests the OnError callback func TestOnError(t *testing.T) { ctx := t.Context() diff --git a/token/services/ttx/finality/mock/tokens_service.go b/token/services/ttx/finality/mock/tokens_service.go index b4578d3e7a..46559aaf05 100644 --- a/token/services/ttx/finality/mock/tokens_service.go +++ b/token/services/ttx/finality/mock/tokens_service.go @@ -10,7 +10,7 @@ import ( ) type TokensService struct { - AppendValidStub func(context.Context, driver.Transaction, token.RequestAnchor, *token.Request) error + AppendValidStub func(context.Context, driver.Transaction, token.RequestAnchor, *token.Request) (func(ctx context.Context), error) appendValidMutex sync.RWMutex appendValidArgsForCall []struct { arg1 context.Context @@ -19,10 +19,12 @@ type TokensService struct { arg4 *token.Request } appendValidReturns struct { - result1 error + result1 func(ctx context.Context) + result2 error } appendValidReturnsOnCall map[int]struct { - result1 error + result1 func(ctx context.Context) + result2 error } GetCachedTokenRequestStub func(string) (*token.Request, []byte) getCachedTokenRequestMutex sync.RWMutex @@ -41,7 +43,7 @@ type TokensService struct { invocationsMutex sync.RWMutex } -func (fake *TokensService) AppendValid(arg1 context.Context, arg2 driver.Transaction, arg3 token.RequestAnchor, arg4 *token.Request) error { +func (fake *TokensService) AppendValid(arg1 context.Context, arg2 driver.Transaction, arg3 token.RequestAnchor, arg4 *token.Request) (func(ctx context.Context), error) { fake.appendValidMutex.Lock() ret, specificReturn := fake.appendValidReturnsOnCall[len(fake.appendValidArgsForCall)] fake.appendValidArgsForCall = append(fake.appendValidArgsForCall, struct { @@ -58,9 +60,9 @@ func (fake *TokensService) AppendValid(arg1 context.Context, arg2 driver.Transac return stub(arg1, arg2, arg3, arg4) } if specificReturn { - return ret.result1 + return ret.result1, ret.result2 } - return fakeReturns.result1 + return fakeReturns.result1, fakeReturns.result2 } func (fake *TokensService) AppendValidCallCount() int { @@ -69,7 +71,7 @@ func (fake *TokensService) AppendValidCallCount() int { return len(fake.appendValidArgsForCall) } -func (fake *TokensService) AppendValidCalls(stub func(context.Context, driver.Transaction, token.RequestAnchor, *token.Request) error) { +func (fake *TokensService) AppendValidCalls(stub func(context.Context, driver.Transaction, token.RequestAnchor, *token.Request) (func(ctx context.Context), error)) { fake.appendValidMutex.Lock() defer fake.appendValidMutex.Unlock() fake.AppendValidStub = stub @@ -82,27 +84,30 @@ func (fake *TokensService) AppendValidArgsForCall(i int) (context.Context, drive return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } -func (fake *TokensService) AppendValidReturns(result1 error) { +func (fake *TokensService) AppendValidReturns(result1 func(ctx context.Context), result2 error) { fake.appendValidMutex.Lock() defer fake.appendValidMutex.Unlock() fake.AppendValidStub = nil fake.appendValidReturns = struct { - result1 error - }{result1} + result1 func(ctx context.Context) + result2 error + }{result1, result2} } -func (fake *TokensService) AppendValidReturnsOnCall(i int, result1 error) { +func (fake *TokensService) AppendValidReturnsOnCall(i int, result1 func(ctx context.Context), result2 error) { fake.appendValidMutex.Lock() defer fake.appendValidMutex.Unlock() fake.AppendValidStub = nil if fake.appendValidReturnsOnCall == nil { fake.appendValidReturnsOnCall = make(map[int]struct { - result1 error + result1 func(ctx context.Context) + result2 error }) } fake.appendValidReturnsOnCall[i] = struct { - result1 error - }{result1} + result1 func(ctx context.Context) + result2 error + }{result1, result2} } func (fake *TokensService) GetCachedTokenRequest(arg1 string) (*token.Request, []byte) { diff --git a/token/services/ttx/finality/recovery_test.go b/token/services/ttx/finality/recovery_test.go index 8009c6960e..a99f48f6c0 100644 --- a/token/services/ttx/finality/recovery_test.go +++ b/token/services/ttx/finality/recovery_test.go @@ -65,7 +65,8 @@ func TestTTXRecoveryHandler_Recover_ValidTransaction_CachedRequest(t *testing.T) mockNetwork.GetTransactionStatusReturns(network.Valid, tokenRequestHash, "", nil) mockTokens.GetCachedTokenRequestReturns(mockRequest, msgToSign) mockTTXDB.NewTransactionReturns(mockTx, nil) - mockTokens.AppendValidReturns(nil) + publishedAfterCommits := -1 + mockTokens.AppendValidReturns(func(context.Context) { publishedAfterCommits = mockTx.CommitCallCount() }, nil) mockTx.SetStatusReturns(nil) mockTx.CommitReturns(nil) mockTTXDB.SetStatusReturns(nil) @@ -82,6 +83,8 @@ func TestTTXRecoveryHandler_Recover_ValidTransaction_CachedRequest(t *testing.T) require.Equal(t, 1, mockTx.SetStatusCallCount()) require.Equal(t, 1, mockTx.CommitCallCount()) require.Equal(t, 0, mockTTXDB.SetStatusCallCount()) + // the token events must be published, and only once the transaction was committed + require.Equal(t, 1, publishedAfterCommits) } func TestTTXRecoveryHandler_Recover_ValidTransaction_LoadFromDB(t *testing.T) { @@ -127,7 +130,8 @@ func TestTTXRecoveryHandler_Recover_ValidTransaction_LoadFromDB(t *testing.T) { mockTTXDB.GetTokenRequestReturns(tokenRequestRaw, nil) mockHasher.ProcessTokenRequestReturns(mockRequest, msgToSign, nil) mockTTXDB.NewTransactionReturns(mockTx, nil) - mockTokens.AppendValidReturns(nil) + publishedAfterCommits := -1 + mockTokens.AppendValidReturns(func(context.Context) { publishedAfterCommits = mockTx.CommitCallCount() }, nil) mockTx.SetStatusReturns(nil) mockTx.CommitReturns(nil) mockTTXDB.SetStatusReturns(nil) @@ -146,6 +150,8 @@ func TestTTXRecoveryHandler_Recover_ValidTransaction_LoadFromDB(t *testing.T) { require.Equal(t, 1, mockTx.SetStatusCallCount()) require.Equal(t, 1, mockTx.CommitCallCount()) require.Equal(t, 0, mockTTXDB.SetStatusCallCount()) + // the token events must be published, and only once the transaction was committed + require.Equal(t, 1, publishedAfterCommits) } func TestTTXRecoveryHandler_Recover_InvalidTransaction(t *testing.T) { @@ -420,7 +426,7 @@ func TestTTXRecoveryHandler_Recover_AppendError(t *testing.T) { mockTokens.GetCachedTokenRequestReturns(mockRequest, msgToSign) mockTTXDB.NewTransactionReturns(mockTx, nil) expectedErr := errors.New("append failed") - mockTokens.AppendValidReturns(expectedErr) + mockTokens.AppendValidReturns(nil, expectedErr) // Execute recoverErr := handler.Recover(ctx, txID)