Skip to content

In passive mode instance do not attempt to send request by GolangLocalRequest, if set AutoRequestSelect #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AndreiShipov merged 1 commit into from
Apr 28, 2026
Merged

In passive mode instance do not attempt to send request by GolangLocalRequest, if set AutoRequestSelect #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions statefun/function_type.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,9 +172,7 @@ func RegisterDynamicFunctionType(runtime *Runtime, name string, logicHandler Fun
}

// Start NATS subscriptions only if this runtime is active and (for single-instance) owns the lock.
runtime.activeInstanceMu.RLock()
isActive := runtime.config.isActiveInstance
runtime.activeInstanceMu.RUnlock()
isActive := runtime.IsActiveInstance()

if isActive && singleInstanceLocked {
if err := ft.startSubscriptions(); err != nil {
Expand Down
3 changes: 2 additions & 1 deletion statefun/io.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,7 +371,8 @@ func (r *Runtime) request(requestProvider sfPlugins.RequestProvider, callerTypen
case sfPlugins.AutoRequestSelect:
selection := sfPlugins.NatsCoreGlobalRequest
if shadowObjectCanBeReceiver || !r.Domain.IsShadowObject(targetID) {
if r.functionTypeIsReadyForGoLangCommunication(targetTypename, true, targetID) == 0 {
if r.functionTypeIsReadyForGoLangCommunication(targetTypename, true, targetID) == 0 &&
r.IsActiveInstance() {
selection = sfPlugins.GolangLocalRequest
}
}
Expand Down
34 changes: 21 additions & 13 deletions statefun/runtime.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ func (r *Runtime) Start(ctx context.Context, cacheConfig *cache.Config) error {
sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
<-sig
if !r.config.isActiveInstance {
if !r.IsActiveInstance() {
logger.Debugf(ctx, "Runtime is not active. Shutting down immediately")
r.gs.beginShutdownPhaseOne()
r.gs.cancelAllContexts()
Expand Down Expand Up @@ -220,7 +220,7 @@ func (r *Runtime) Start(ctx context.Context, cacheConfig *cache.Config) error {

<-r.Domain.cache.Synced

if r.config.isActiveInstance {
if r.IsActiveInstance() {
logger.Debugf(ctx, "Shutdown - waiting for transaction committer")
<-r.Domain.shutdown
}
Expand Down Expand Up @@ -253,7 +253,7 @@ func (r *Runtime) Start(ctx context.Context, cacheConfig *cache.Config) error {
if err != nil {
if errors.Is(err, ErrMutexLocked) {
logger.Debugf(ctx, "Cant lock. Another runtime is already active")
r.config.isActiveInstance = false
r.setActiveInstance(false)
} else {
return err
}
Expand All @@ -264,11 +264,11 @@ func (r *Runtime) Start(ctx context.Context, cacheConfig *cache.Config) error {
}()
}
} else {
r.config.isActiveInstance = true
r.setActiveInstance(true)
}

// if active - can publish to WAL, passive - can not
r.Domain.Cache().SetWALWriteEnabled(r.config.isActiveInstance)
r.Domain.Cache().SetWALWriteEnabled(r.IsActiveInstance())

// Handle single-instance functions.
singleInstanceFunctionRevisions := make(map[string]uint64)
Expand All @@ -277,7 +277,7 @@ func (r *Runtime) Start(ctx context.Context, cacheConfig *cache.Config) error {
}

// Start function subscriptions.
if r.config.isActiveInstance {
if r.IsActiveInstance() {
if err := r.startFunctionSubscriptions(ctx, singleInstanceFunctionRevisions); err != nil {
return err
}
Expand Down Expand Up @@ -619,7 +619,7 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str

//release all functions
releaseAllLocks := func(ctx context.Context, runtime *Runtime, revisions map[string]uint64) {
if runtime.config.isActiveInstance {
if runtime.IsActiveInstance() {
for ftName, revID := range revisions {
system.MsgOnErrorReturn(KeyMutexUnlock(ctx, runtime, system.GetHashStr(ftName), revID))
}
Expand All @@ -633,7 +633,7 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str
becomePassive := func(cause string) {
lg.Logf(lg.WarnLevel, "%s, becoming passive", cause)
r.activeInstanceMu.Lock()
r.config.isActiveInstance = false
r.setActiveInstance(false)
r.config.activeRevID = 0
r.activeInstanceMu.Unlock()
r.Domain.Cache().SetWALWriteEnabled(false)
Expand Down Expand Up @@ -665,7 +665,7 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str
r.config.activeRevID = newRevID
}

if r.config.isActiveInstance {
if r.IsActiveInstance() {
// Already active — nothing to do
} else if kvConsistencyCheck != nil {
// Activating: consistency check in progress, lock is being refreshed above
Expand All @@ -676,13 +676,13 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str
lg.Logf(lg.ErrorLevel, "KV consistency check failed: %v", err)
system.MsgOnErrorReturn(KeyMutexUnlock(ctx, r, system.GetHashStr(RuntimeName), r.config.activeRevID))
r.activeInstanceMu.Lock()
r.config.isActiveInstance = false
r.setActiveInstance(false)
r.config.activeRevID = 0
r.activeInstanceMu.Unlock()
} else {
lg.Logf(lg.DebugLevel, "KV consistent, fully active now")
r.activeInstanceMu.Lock()
r.config.isActiveInstance = true
r.setActiveInstance(true)
r.activeInstanceMu.Unlock()
r.Domain.Cache().SetWALWriteEnabled(true)
r.gs.resetPhaseOneCtx()
Expand All @@ -708,7 +708,7 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str
}
}
} else {
r.config.isActiveInstance = true
r.setActiveInstance(true)
}

tryLock := func(ftName string) {
Expand All @@ -720,7 +720,7 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str
}
}

if r.config.isActiveInstance {
if r.IsActiveInstance() {
for ftName, revID := range revisions {
if revID == 0 {
tryLock(ftName)
Expand Down Expand Up @@ -757,9 +757,17 @@ func (r *Runtime) runtimeLifecycleUpdater(ctx context.Context, revisions map[str
// IsActiveInstance indicates whether this runtime instance currently owns
// the active role in HA mode
func (r *Runtime) IsActiveInstance() bool {
r.activeInstanceMu.RLock()
defer r.activeInstanceMu.RUnlock()
return r.config.isActiveInstance
}

func (r *Runtime) setActiveInstance(active bool) {
r.activeInstanceMu.Lock()
defer r.activeInstanceMu.Unlock()
r.config.isActiveInstance = active
}

// contains checks if a slice contains a particular string.
func contains(slice []string, item string) bool {
for _, s := range slice {
Expand Down
Loading