add an option to disable the deadline timeout for the FSM broadcast (#19)

* add an option to disable the deadline timeout for the FSM broadcast

Author: robbyt

README.md

@@ -134,8 +134,18 @@ currentState := machine.GetState()
 
 ### State Change Notifications
 
+#### Broadcast Modes
+
+Subscribers can operate in three different modes based on their timeout setting:
+
+**Async Mode (timeout=0)**: State updates are dropped if the channel is full. Non-blocking transitions. This is the default behavior.
+
+**Sync Mode (positive timeout)**: Blocks state transitions until all sync subscribers read the update or timeout. Never drops state updates unless timeout is reached.
+
+**Infinite Blocking Mode (negative timeout)**: Blocks state transitions indefinitely until all infinite subscribers read the update. Never drops state updates or times out.
+
 ```go
-// Get notification channel
+// Get notification channel with default async behavior (timeout=0)
 ctx, cancel := context.WithCancel(context.Background())
 defer cancel()
 
@@ -149,18 +159,23 @@ go func() {
 	}
 }()
 
-// Alternative: Add a direct subscriber channel
-stateCh := make(chan string, 1)
-unsubscribe := machine.AddSubscriber(stateCh)
+// Use sync broadcast with a 10s timeout (WithSyncBroadcast is a shortcut for settings a 10s timeout)
+syncChan := machine.GetStateChanWithOptions(ctx, fsm.WithSyncBroadcast())
 
-// Process state updates in a separate goroutine
+// Use sync broadcast with 1hr custom timeout
+timeoutChan := machine.GetStateChanWithOptions(ctx, fsm.WithSyncTimeout(1*time.Hour))
+
+// Use infinite blocking (never times out)
+infiniteChan := machine.GetStateChanWithOptions(ctx,
+	fsm.WithSyncTimeout(-1),
+)
+
+// Read and print all state changes from the channel
 go func() {
-	for state := range stateCh {
-		fmt.Println("State updated:", state)
+	for state := range syncChan {
+		fmt.Println("State:", state)
 	}
 }()
-
-defer unsubscribe() // Call to stop receiving updates
 ```
 
 ## Complete Example

stateChan.go

@@ -23,7 +23,8 @@ import (
 )
 
 const (
-	defaultSyncTimeout = 10 * time.Second
+	defaultAsyncTimeout = 0
+	defaultSyncTimeout  = 10 * time.Second
 )
 
 // SubscriberOption configures subscriber behavior
@@ -33,7 +34,6 @@ type SubscriberOption func(*subscriberConfig)
 type subscriberConfig struct {
 	sendInitial   bool
 	customChannel chan string
-	syncBroadcast bool
 	syncTimeout   time.Duration
 }
 
@@ -62,7 +62,7 @@ func WithCustomChannel(ch chan string) SubscriberOption {
 // is delivered to the channel, rather than dropping messages for full channels
 func WithSyncBroadcast() SubscriberOption {
 	return func(config *subscriberConfig) {
-		config.syncBroadcast = true
+		config.syncTimeout = defaultSyncTimeout
 	}
 }
 
@@ -85,7 +85,7 @@ func (fsm *Machine) GetStateChanWithOptions(
 ) <-chan string {
 	config := &subscriberConfig{
 		sendInitial: true,
-		syncTimeout: defaultSyncTimeout,
+		syncTimeout: defaultAsyncTimeout,
 	}
 
 	for _, opt := range opts {
@@ -147,7 +147,7 @@ func (fsm *Machine) GetStateChanBuffer(ctx context.Context, chanBufferSize int)
 func (fsm *Machine) AddSubscriber(ch chan string) func() {
 	config := &subscriberConfig{
 		sendInitial: true,
-		syncTimeout: defaultSyncTimeout,
+		syncTimeout: defaultAsyncTimeout,
 	}
 	return fsm.addSubscriberWithConfig(ch, config)
 }
@@ -183,8 +183,9 @@ func (fsm *Machine) unsubscribe(ch chan string) {
 }
 
 // broadcast sends the new state to all subscriber channels.
-// For async subscribers, if a channel is full, the state change is skipped for that channel, and a warning is logged.
-// For sync subscribers, the broadcast blocks until the message is delivered or times out after 10 seconds.
+// Subscribers with negative timeout block indefinitely until delivered.
+// Subscribers with timeout=0 behave asynchronously (drop messages if channel is full).
+// Subscribers with positive timeout block until delivered or timeout.
 // This, and the other subscriber-related methods, use a standard mutex instead of an RWMutex,
 // because the broadcast sends should always be serial, and never concurrent, otherwise the order
 // of state change notifications could be unpredictable.
@@ -208,21 +209,15 @@ func (fsm *Machine) broadcast(state string) {
 			return true
 		}
 
-		if config.syncBroadcast {
-			// Handle sync broadcast with timeout in parallel goroutines
+		if config.syncTimeout < 0 {
+			// Handle infinite blocking broadcast
 			wg.Add(1)
 			go func(ch chan string) {
 				defer wg.Done()
-				select {
-				case ch <- state:
-					logger.Debug("State delivered to synchronous subscriber")
-				case <-time.After(config.syncTimeout):
-					logger.Warn("Synchronous subscriber blocked; state delivery timed out",
-						"timeout", config.syncTimeout,
-						"channel_capacity", cap(ch), "channel_length", len(ch))
-				}
+				ch <- state
+				logger.Debug("State delivered to blocking subscriber")
 			}(ch)
-		} else {
+		} else if config.syncTimeout == 0 {
 			// Handle async broadcast (non-blocking)
 			select {
 			case ch <- state:
@@ -231,6 +226,20 @@ func (fsm *Machine) broadcast(state string) {
 				logger.Debug("Asynchronous subscriber channel full; state delivery skipped",
 					"channel_capacity", cap(ch), "channel_length", len(ch))
 			}
+		} else {
+			// Handle sync broadcast with positive timeout
+			wg.Add(1)
+			go func(ch chan string, timeout time.Duration) {
+				defer wg.Done()
+				select {
+				case ch <- state:
+					logger.Debug("State delivered to synchronous subscriber")
+				case <-time.After(timeout):
+					logger.Warn("Synchronous subscriber blocked; state delivery timed out",
+						"timeout", timeout,
+						"channel_capacity", cap(ch), "channel_length", len(ch))
+				}
+			}(ch, config.syncTimeout)
 		}
 		return true // continue iteration
 	})

stateChan_test.go

@@ -1139,4 +1139,44 @@ func TestFSM_GetStateChanWithOptions(t *testing.T) {
 			}
 		}, time.Second, 10*time.Millisecond, "Async subscriber should receive state despite blocked sync subscriber")
 	})
+
+	t.Run("Negative timeout blocks indefinitely", func(t *testing.T) {
+		if testing.Short() {
+			t.Skip("Skipping in short mode")
+		}
+
+		fsm, err := New(nil, StatusNew, TypicalTransitions)
+		require.NoError(t, err)
+		ctx := t.Context()
+
+		// Create a permanently blocked sync subscriber with negative timeout
+		blockedSyncCh := make(chan string) // Unbuffered, no readers
+		fsm.GetStateChanWithOptions(ctx,
+			WithSyncTimeout(-1), // Negative timeout should block indefinitely
+			WithCustomChannel(blockedSyncCh),
+			WithoutInitialState(),
+		)
+
+		// Create a normal async subscriber that should not be blocked
+		asyncCh := fsm.GetStateChanWithOptions(ctx, WithBufferSize(2))
+		<-asyncCh // Consume initial state
+
+		// Start transition in a goroutine since it should block indefinitely
+		transitionDone := make(chan struct{})
+		go func() {
+			defer close(transitionDone)
+			err := fsm.Transition(StatusBooting)
+			require.NoError(t, err)
+		}()
+
+		// Transition should never complete within 11 seconds due to negative timeout
+		assert.Never(t, func() bool {
+			select {
+			case <-transitionDone:
+				return true
+			default:
+				return false
+			}
+		}, 11*time.Second, 100*time.Millisecond, "Transition should not complete with negative timeout blocking indefinitely")
+	})
 }