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 <hayimsha@fhe03.vpc.cloud9.ibm.com>

Author: Hayim.Shaul@ibm.com

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