Merge pull request #1729 from lightninglabs/sync-supply-commit

universe/supplycommit: add done channel to enable sync processing of events

Author: Roasbeef

rpcserver.go

@@ -4085,7 +4085,9 @@ func (r *rpcServer) IgnoreAssetOutPoint(ctx context.Context,
 	ignoreEvent := supplycommit.NewIgnoreEvent{
 		SignedIgnoreTuple: signedIgnore,
 	}
-	err = r.cfg.SupplyCommitManager.SendEvent(ctx, assetSpec, &ignoreEvent)
+	err = r.cfg.SupplyCommitManager.SendEventSync(
+		ctx, assetSpec, &ignoreEvent,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to upsert ignore tuple: %w", err)
 	}

universe/supplycommit/multi_sm_manager.go

@@ -225,6 +225,46 @@ func (m *MultiStateMachineManager) SendEvent(ctx context.Context,
 	return nil
 }
 
+// SendEventSync sends an event to the state machine and waits for it to be
+// processed and written to disk. This method provides synchronous confirmation
+// that the event has been durably persisted. If the event doesn't support
+// synchronous processing (i.e., it's not a SupplyUpdateEvent), this method will
+// return an error.
+func (m *MultiStateMachineManager) SendEventSync(ctx context.Context,
+	assetSpec asset.Specifier, event SyncSupplyUpdateEvent) error {
+
+	// Only SupplyUpdateEvents can be processed synchronously.
+	supplyEvent, ok := event.(SupplyUpdateEvent)
+	if !ok {
+		return fmt.Errorf("event type %T does not support "+
+			"synchronous processing", event)
+	}
+
+	// We'll use this channel to signal when the event has been processed.
+	done := make(chan error, 1)
+
+	// As the event has already been created, we'll set the done channel
+	// directly.
+	switch e := supplyEvent.(type) {
+	case *NewMintEvent:
+		e.Done = done
+	case *NewBurnEvent:
+		e.Done = done
+	case *NewIgnoreEvent:
+		e.Done = done
+	default:
+		return fmt.Errorf("unexpected supply update event type: %T",
+			supplyEvent)
+	}
+
+	// Send the event to the state machine, the pause and wait until it
+	// signals that the event has been fully processed.
+	if err := m.SendEvent(ctx, assetSpec, supplyEvent); err != nil {
+		return err
+	}
+	return event.WaitForDone(ctx)
+}
+
 // CanHandle determines if the state machine associated with the given asset
 // specifier can handle the given message. If a state machine for the asset
 // group does not exist, it will be created and started.

universe/supplycommit/state_machine_test.go

@@ -93,6 +93,7 @@ func newTestMintEvent(t *testing.T, scriptKey *btcec.PublicKey,
 		LeafKey:       leafKey,
 		IssuanceProof: issuanceProof,
 		MintHeight:    1000,
+		Done:          make(chan error, 1),
 	}
 }
 
@@ -326,6 +327,32 @@ func (h *supplyCommitTestHarness) sendEvent(event Event) {
 	h.t.Helper()
 
 	h.stateMachine.SendEvent(context.Background(), event)
+
+	// If this is a SyncSupplyUpdateEvent with a done channel, wait for it.
+	if syncEvent, ok := event.(SyncSupplyUpdateEvent); ok {
+		// Check if the event has a done channel by attempting to wait.
+		// We use a short timeout to avoid blocking indefinitely.
+		ctx, cancel := context.WithTimeout(
+			context.Background(), testTimeout,
+		)
+		defer cancel()
+
+		err := syncEvent.WaitForDone(ctx)
+
+		switch {
+		// If the done channel is nil, this is async mode - not an
+		// error.
+		case errors.Is(err, ErrNilDoneChannel):
+			return
+
+		// If we got a timeout, that's a real error.
+		case errors.Is(err, ErrEventTimeout):
+			require.NoError(h.t, err, "sync event timed out")
+		}
+
+		// For other errors (e.g., database errors), tests should verify
+		// them through other means (e.g., assertNoStateTransitions).
+	}
 }
 
 func (h *supplyCommitTestHarness) expectInsertPendingUpdate(
@@ -1418,7 +1445,9 @@ func TestSupplyUpdateEventTypes(t *testing.T) {
 
 		// Finally, we assert that the RawProof field of the decoded
 		// event matches the original serialized proof bytes, confirming
-		// a successful round trip.
+		// a successful round trip. Note: We don't encode/decode the Done
+		// channel, so we clear it before comparison.
+		originalMintEvent.Done = nil
 		require.Equal(t, originalMintEvent, decodedMintEvent)
 
 		require.Equal(
@@ -1903,3 +1932,81 @@ func TestStateAndEventMethods(t *testing.T) {
 		require.False(t, (&DefaultState{}).IsTerminal())
 	})
 }
+
+// TestSyncSupplyUpdateEventEdgeCases tests edge cases for synchronous event
+// handling.
+func TestSyncSupplyUpdateEventEdgeCases(t *testing.T) {
+	t.Parallel()
+
+	testScriptKey := test.RandPubKey(t)
+	defaultAssetSpec := asset.NewSpecifierFromId(testAssetID)
+
+	// Test that if we get an error during processing, it propagates..
+	t.Run("error_propagation", func(t *testing.T) {
+		h := newSupplyCommitTestHarness(t, &harnessCfg{
+			initialState: &DefaultState{},
+			assetSpec:    defaultAssetSpec,
+		})
+		h.start()
+		defer h.stopAndAssert()
+
+		mintEvent := newTestMintEvent(
+			t, testScriptKey, randOutPoint(t),
+		)
+
+		// Expect the insert to fail.
+		insertErr := errors.New("database error")
+		h.mockStateLog.On(
+			"InsertPendingUpdate", mock.Anything, h.cfg.assetSpec,
+			mintEvent,
+		).Return(insertErr).Once()
+
+		h.expectFailure(insertErr)
+		h.sendEvent(mintEvent)
+
+		// The sendEvent method should have already waited and checked
+		// the error. We're verifying that the test harness correctly
+		// propagated the error.
+		h.assertNoStateTransitions()
+	})
+
+	// If we cancel the context, WaitForDone should return an error.
+	t.Run("context_cancellation", func(t *testing.T) {
+		mintEvent := newTestMintEvent(
+			t, testScriptKey, randOutPoint(t),
+		)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		err := mintEvent.WaitForDone(ctx)
+		require.Error(t, err, "expected context cancellation error")
+		require.ErrorIs(t, err, ErrEventTimeout)
+		require.ErrorIs(t, err, context.Canceled)
+	})
+
+	// Passing a nil done channel should return an error.
+	t.Run("nil_done_channel", func(t *testing.T) {
+		mintEvent := &NewMintEvent{
+			LeafKey: universe.AssetLeafKey{
+				BaseLeafKey: universe.BaseLeafKey{
+					OutPoint: randOutPoint(t),
+					ScriptKey: &asset.ScriptKey{
+						PubKey: testScriptKey,
+					},
+				},
+				AssetID: testAssetID,
+			},
+			IssuanceProof: universe.Leaf{
+				GenesisWithGroup: universe.GenesisWithGroup{
+					Genesis: testGenesis,
+				},
+			},
+			MintHeight: 1000,
+		}
+
+		err := mintEvent.WaitForDone(context.Background())
+		require.Error(t, err, "expected error for nil done channel")
+		require.ErrorIs(t, err, ErrNilDoneChannel)
+	})
+}

universe/supplycommit/states.go

@@ -2,6 +2,7 @@ package supplycommit
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 
@@ -20,6 +21,14 @@ var (
 	// ErrInvalidStateTransition is returned when we receive an unexpected
 	// event for a given state.
 	ErrInvalidStateTransition = fmt.Errorf("invalid state transition")
+
+	// ErrEventTimeout is returned when waiting for a synchronous event
+	// times out due to context cancellation.
+	ErrEventTimeout = fmt.Errorf("event processing timeout")
+
+	// ErrNilDoneChannel is returned when attempting to wait for a
+	// synchronous event that doesn't have a done channel.
+	ErrNilDoneChannel = fmt.Errorf("done channel is nil")
 )
 
 // Event is a special interface used to create the equivalent of a sum-type, but
@@ -96,15 +105,68 @@ type SupplyUpdateEvent interface {
 	Encode(io.Writer) error
 }
 
+// SyncSupplyUpdateEvent is an interface that extends SupplyUpdateEvent with
+// the ability to signal completion when the event has been processed.
+type SyncSupplyUpdateEvent interface {
+	SupplyUpdateEvent
+
+	// SignalDone signals completion on the done channel if it exists.
+	// It sends the error (or nil for success) and does not block.
+	SignalDone(error)
+
+	// WaitForDone waits for the event to be processed by waiting on its
+	// done channel. This is a helper method that can be used by callers who
+	// want synchronous confirmation that their event has been persisted.
+	WaitForDone(context.Context) error
+}
+
+// waitForDone is an internal helper function that waits for an event to be
+// processed by waiting on its done channel.
+func waitForDone(ctx context.Context, done chan error) error {
+	if done == nil {
+		return ErrNilDoneChannel
+	}
+
+	select {
+	case err := <-done:
+		return err
+	case <-ctx.Done():
+		return fmt.Errorf("%w: %w", ErrEventTimeout, ctx.Err())
+	}
+}
+
 // NewIgnoreEvent signals that a caller wishes to update the ignore portion of
 // the supply tree with a new outpoint + script key combo.
 type NewIgnoreEvent struct {
 	universe.SignedIgnoreTuple
+
+	// Done is an optional channel that will receive an error (or nil for
+	// success) when the event has been processed and written to disk. If
+	// nil, the event is processed asynchronously.
+	Done chan error
 }
 
 // eventSealed is a special method that is used to seal the interface.
 func (n *NewIgnoreEvent) eventSealed() {}
 
+// SignalDone signals completion on the done channel if it exists. It sends the
+// error (or nil for success) and does not block.
+func (n *NewIgnoreEvent) SignalDone(err error) {
+	if n.Done != nil {
+		select {
+		case n.Done <- err:
+		default:
+		}
+	}
+}
+
+// WaitForDone waits for the event to be processed by waiting on its done
+// channel. This is a helper method that can be used by callers who want
+// synchronous confirmation that their event has been persisted.
+func (n *NewIgnoreEvent) WaitForDone(ctx context.Context) error {
+	return waitForDone(ctx, n.Done)
+}
+
 // BlockHeight returns the block height of the update.
 func (n *NewIgnoreEvent) BlockHeight() uint32 {
 	return n.SignedIgnoreTuple.IgnoreTuple.Val.BlockHeight
@@ -141,15 +203,42 @@ func (n *NewIgnoreEvent) Encode(w io.Writer) error {
 // SupplyUpdateEvent interface.
 var _ SupplyUpdateEvent = (*NewIgnoreEvent)(nil)
 
+// A compile time assertion to ensure that NewIgnoreEvent implements the
+// SyncSupplyUpdateEvent interface.
+var _ SyncSupplyUpdateEvent = (*NewIgnoreEvent)(nil)
+
 // NewBurnEvent signals that a caller wishes to update the burn portion of
 // the supply tree with a new burnt asset.
 type NewBurnEvent struct {
 	universe.BurnLeaf
+
+	// Done is an optional channel that will receive an error (or nil for
+	// success) when the event has been processed and written to disk.
+	// If nil, the event is processed asynchronously.
+	Done chan error
 }
 
 // eventSealed is a special method that is used to seal the interface.
 func (n *NewBurnEvent) eventSealed() {}
 
+// SignalDone signals completion on the done channel if it exists. It sends
+// the error (or nil for success) and does not block.
+func (n *NewBurnEvent) SignalDone(err error) {
+	if n.Done != nil {
+		select {
+		case n.Done <- err:
+		default:
+		}
+	}
+}
+
+// WaitForDone waits for the event to be processed by waiting on its done
+// channel. This is a helper method that can be used by callers who want
+// synchronous confirmation that their event has been persisted.
+func (n *NewBurnEvent) WaitForDone(ctx context.Context) error {
+	return waitForDone(ctx, n.Done)
+}
+
 // BlockHeight returns the block height of the update.
 func (n *NewBurnEvent) BlockHeight() uint32 {
 	return n.BurnLeaf.BurnProof.BlockHeight
@@ -187,6 +276,10 @@ func (n *NewBurnEvent) Encode(w io.Writer) error {
 // SupplyUpdateEvent interface.
 var _ SupplyUpdateEvent = (*NewBurnEvent)(nil)
 
+// A compile time assertion to ensure that NewBurnEvent implements the
+// SyncSupplyUpdateEvent interface.
+var _ SyncSupplyUpdateEvent = (*NewBurnEvent)(nil)
+
 // NewMintEvent signals that a caller wishes to update the mint portion of the
 // supply tree with a new minted asset.
 type NewMintEvent struct {
@@ -198,6 +291,11 @@ type NewMintEvent struct {
 
 	// MintHeight is the height of the block that contains the mint.
 	MintHeight uint32
+
+	// Done is an optional channel that will receive an error (or nil for
+	// success) when the event has been processed and written to disk.
+	// If nil, the event is processed asynchronously.
+	Done chan error
 }
 
 // BlockHeight returns the block height of the update.
@@ -208,6 +306,24 @@ func (n *NewMintEvent) BlockHeight() uint32 {
 // eventSealed is a special method that is used to seal the interface.
 func (n *NewMintEvent) eventSealed() {}
 
+// SignalDone signals completion on the done channel if it exists. It sends
+// the error (or nil for success) and does not block.
+func (n *NewMintEvent) SignalDone(err error) {
+	if n.Done != nil {
+		select {
+		case n.Done <- err:
+		default:
+		}
+	}
+}
+
+// WaitForDone waits for the event to be processed by waiting on its done
+// channel. This is a helper method that can be used by callers who want
+// synchronous confirmation that their event has been persisted.
+func (n *NewMintEvent) WaitForDone(ctx context.Context) error {
+	return waitForDone(ctx, n.Done)
+}
+
 // ScriptKey returns the script key that is used to identify the target
 // asset.
 func (n *NewMintEvent) ScriptKey() asset.SerializedKey {
@@ -279,6 +395,10 @@ func (n *NewMintEvent) Decode(r io.Reader) error {
 // SupplyUpdateEvent interface.
 var _ SupplyUpdateEvent = (*NewMintEvent)(nil)
 
+// A compile time assertion to ensure that NewMintEvent implements the
+// SyncSupplyUpdateEvent interface.
+var _ SyncSupplyUpdateEvent = (*NewMintEvent)(nil)
+
 // DefaultState is the idle state of the state machine. We start in this state
 // when there are no pending changes that need to committed.
 //