fix(tokens): publish token events only after commit

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
 

token/services/tokens/storage.go

@@ -116,13 +116,22 @@ 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
 	// Tx is the underlying database transaction.
 	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.

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) {