fix HA

statefun/runtime.go

@@ -312,7 +312,7 @@
 	wg.Wait()
 }
 
 func (r *Runtime) stopRequestSubscriptions() {
 	for {
 		allFunctionsReadyForShutdown := true
 		for _, ft := range r.registeredFunctionTypes {
@@ -605,7 +605,6 @@
 		r.stopFunctionSubscriptions(ctx)
 		if r.afterStartRunning.Load() {
 			r.gs.cancelPhaseOne()
-			r.afterStartRunning.Store(false)
 		}
 		kvConsistencyCheck = nil
 	}
@@ -650,6 +649,7 @@
 							r.config.isActiveInstance = true
 							r.activeInstanceMu.Unlock()
 							r.gs.resetPhaseOneCtx()
+							r.afterStartRunning.Store(false)
 							subscribeRequired = true
 						}
 					default:
@@ -684,11 +684,6 @@
 			}
 
 			if r.config.isActiveInstance {
-				if r.afterStartRunning.CompareAndSwap(false, true) {
-					lg.GetLogger().Debugf(ctx, "runtime is active, run afterStartFunctions")
-					r.runAfterStartFunctions(r.gs.phaseOneCtx())
-				}
-
 				for ftName, revID := range revisions {
 					if revID == 0 {
 						tryLock(ftName)
@@ -713,6 +708,11 @@
 					lg.Logf(lg.ErrorLevel, "function subscriptions failed: %v", err)
 				}
 			}
+
+			if r.afterStartRunning.CompareAndSwap(false, true) {
+				lg.GetLogger().Debugf(ctx, "run afterStartFunctions")
+				r.runAfterStartFunctions(r.gs.phaseOneCtx())
+			}
 		}
 	}
 }

statefun/wal.go

@@ -15,6 +15,8 @@ import (
 	"github.com/nats-io/nats.go"
 )
 
+var errTransactionAlreadyApplied = fmt.Errorf("transaction already applied by another runtime")
+
 const (
 	WALOperationsStreamName = "wal_operations"
 	WALCommitsStreamName    = "wal_commits"
@@ -136,6 +138,12 @@ func (dm *Domain) runTransactionCommitter(ctx context.Context, ready chan struct
 			}
 
 			if err := dm.applyTransactionOperations(ctx, txID); err != nil {
+				if errors.Is(err, errTransactionAlreadyApplied) {
+					lg.Logf(lg.DebugLevel, "TransactionCommitter: transaction %s already applied, acking", txID)
+					system.MsgOnErrorReturn(msg.Ack())
+					processedCount++
+					return
+				}
 				lg.Logf(lg.ErrorLevel, "TransactionCommitter: failed to apply transaction %s: %s", txID, err)
 				system.MsgOnErrorReturn(msg.Nak())
 				return
@@ -237,6 +245,10 @@ func (dm *Domain) applyTransactionOperations(ctx context.Context, txID string) e
 
 	sub, err := dm.js.PullSubscribe(opsSubject, consumerName)
 	if err != nil {
+		if errors.Is(err, nats.ErrConsumerDeleted) {
+			lg.Logf(lg.DebugLevel, "Per-tx consumer %s already deleted, transaction applied by another runtime", consumerName)
+			return errTransactionAlreadyApplied
+		}
 		lg.Logf(lg.ErrorLevel, "Failed to create subscription: %s", err)
 		return fmt.Errorf("failed to subscribe to operations: %w", err)
 	}
@@ -258,6 +270,10 @@ func (dm *Domain) applyTransactionOperations(ctx context.Context, txID string) e
 				lg.Logf(lg.TraceLevel, "applyTransactionOperations: finished, processed %d operations for tx_id=%s", totalOps, txID)
 				break
 			}
+			if errors.Is(err, nats.ErrConsumerDeleted) {
+				lg.Logf(lg.DebugLevel, "Per-tx consumer deleted during fetch, transaction applied by another runtime (tx_id=%s)", txID)
+				return errTransactionAlreadyApplied
+			}
 			return fmt.Errorf("failed to fetch operations: %w", err)
 		}