diff --git a/backend/cmd/server/servicemanager.go b/backend/cmd/server/servicemanager.go index 2daa763dba..dc6511fc55 100644 --- a/backend/cmd/server/servicemanager.go +++ b/backend/cmd/server/servicemanager.go @@ -548,7 +548,8 @@ func initSessionService(ctx context.Context, svc serverconfig.ServerConfigServic criteriaRevoker flowsession.CriteriaRevoker, logger *log.Logger) (flowsession.Service, flowsession.Config) { cfg := readSessionConfig(ctx, svc, logger) sessionService, err := flowsession.Initialize(dbprovider.GetDBProvider(), deploymentID, - flowsession.NewTimeouts(cfg.IdleTimeoutSeconds, cfg.AbsoluteTimeoutSeconds), criteriaRevoker) + flowsession.NewTimeouts(cfg.IdleTimeoutSeconds, cfg.AbsoluteTimeoutSeconds, + cfg.ActivityRefreshIntervalSeconds), criteriaRevoker) fatalOnError(ctx, logger, err, "Failed to initialize SSO session service") return sessionService, cfg } diff --git a/backend/internal/flow/flowexec/init.go b/backend/internal/flow/flowexec/init.go index 393399876d..1dd745eff1 100644 --- a/backend/internal/flow/flowexec/init.go +++ b/backend/internal/flow/flowexec/init.go @@ -59,7 +59,8 @@ func Initialize( // bound its lifetime to the session's configured absolute timeout (same fallback as the session // executor's timeouts). ssoTransport := session.NewCookieTransport(cfg.SecureCookies) - sessionTimeouts := session.NewTimeouts(cfg.Session.IdleTimeoutSeconds, cfg.Session.AbsoluteTimeoutSeconds) + sessionTimeouts := session.NewTimeouts(cfg.Session.IdleTimeoutSeconds, cfg.Session.AbsoluteTimeoutSeconds, + cfg.Session.ActivityRefreshIntervalSeconds) handler := newFlowExecutionHandler(flowExecService, ssoTransport, sessionTimeouts.Absolute) registerRoutes(mux, handler) diff --git a/backend/internal/flow/session/config.go b/backend/internal/flow/session/config.go index dc298ad3b2..0b3fe93b12 100644 --- a/backend/internal/flow/session/config.go +++ b/backend/internal/flow/session/config.go @@ -29,10 +29,17 @@ import ( type Config struct { IdleTimeoutSeconds int64 `json:"idleTimeoutSeconds" yaml:"idleTimeoutSeconds"` AbsoluteTimeoutSeconds int64 `json:"absoluteTimeoutSeconds" yaml:"absoluteTimeoutSeconds"` + // ActivityRefreshIntervalSeconds is the minimum spacing between persisted activity refreshes: a + // session reuse within this window of the last persisted activity refresh skips the idle-slide + // write, cutting write load on the hot path. It is honored as configured and must be less than + // idleTimeoutSeconds (enforced by Validate), so an active session is never skipped past its idle + // deadline. + ActivityRefreshIntervalSeconds int64 `json:"activityRefreshIntervalSeconds" yaml:"activityRefreshIntervalSeconds"` } -// Validate ensures the configured session timeouts are coherent. Unset (zero) values are allowed -// and fall back to defaults, so only set values are checked. +// Validate ensures the configured session timeouts are coherent. Unset (zero) values are allowed and +// fall back to defaults; the activity-refresh/idle invariant is checked against the resolved +// durations so a defaulted side cannot silently violate it. func (c Config) Validate() error { if c.IdleTimeoutSeconds < 0 { return fmt.Errorf("session.idleTimeoutSeconds must be greater than or equal to 0") @@ -40,10 +47,28 @@ func (c Config) Validate() error { if c.AbsoluteTimeoutSeconds < 0 { return fmt.Errorf("session.absoluteTimeoutSeconds must be greater than or equal to 0") } + if c.ActivityRefreshIntervalSeconds < 0 { + return fmt.Errorf("session.activityRefreshIntervalSeconds must be greater than or equal to 0") + } + // Reject values that would overflow when converted to a time.Duration (nanoseconds); such a value + // wraps to a negative duration and would otherwise slip past the invariant check below. + if c.IdleTimeoutSeconds > maxTimeoutSeconds || c.AbsoluteTimeoutSeconds > maxTimeoutSeconds || + c.ActivityRefreshIntervalSeconds > maxTimeoutSeconds { + return fmt.Errorf("session timeout seconds must not exceed %d", maxTimeoutSeconds) + } if c.IdleTimeoutSeconds > 0 && c.AbsoluteTimeoutSeconds > 0 && c.IdleTimeoutSeconds > c.AbsoluteTimeoutSeconds { return fmt.Errorf("session.idleTimeoutSeconds must not exceed absoluteTimeoutSeconds") } + // Check the resolved durations, not the raw config values: a zero means "use the default", so the + // activity-refresh/idle invariant must be verified after defaults are substituted (and idle is + // clamped to absolute). Comparing only raw positive values lets a defaulted side violate it, e.g. + // a small configured idle with a defaulted refresh, or a defaulted idle with a large configured + // refresh, either of which could skip an active session past its idle deadline. + resolved := NewTimeouts(c.IdleTimeoutSeconds, c.AbsoluteTimeoutSeconds, c.ActivityRefreshIntervalSeconds) + if resolved.ActivityRefresh >= resolved.Idle { + return fmt.Errorf("session.activityRefreshIntervalSeconds must be less than idleTimeoutSeconds") + } return nil } @@ -82,5 +107,8 @@ func (ConfigHandler) Merge(readOnly, writable any) any { if wr.AbsoluteTimeoutSeconds > 0 { merged.AbsoluteTimeoutSeconds = wr.AbsoluteTimeoutSeconds } + if wr.ActivityRefreshIntervalSeconds > 0 { + merged.ActivityRefreshIntervalSeconds = wr.ActivityRefreshIntervalSeconds + } return merged } diff --git a/backend/internal/flow/session/config_test.go b/backend/internal/flow/session/config_test.go index c166e975c8..ccd475d3b9 100644 --- a/backend/internal/flow/session/config_test.go +++ b/backend/internal/flow/session/config_test.go @@ -50,6 +50,41 @@ func (s *ConfigTestSuite) TestValidate_IdleExceedsAbsolute() { s.Require().Error(Config{IdleTimeoutSeconds: 28801, AbsoluteTimeoutSeconds: 28800}.Validate()) } +func (s *ConfigTestSuite) TestValidate_OverflowingSecondsRejected() { + // A second value beyond what a time.Duration can hold would wrap negative; it must be rejected + // rather than slip past the invariant check. + s.Require().Error(Config{IdleTimeoutSeconds: maxTimeoutSeconds + 1}.Validate()) + s.Require().Error(Config{AbsoluteTimeoutSeconds: maxTimeoutSeconds + 1}.Validate()) + s.Require().Error(Config{ActivityRefreshIntervalSeconds: maxTimeoutSeconds + 1}.Validate()) +} + +func (s *ConfigTestSuite) TestValidate_ActivityRefreshInterval() { + // An activity-refresh interval below the idle window is accepted. + s.Require().NoError(Config{IdleTimeoutSeconds: 1800, ActivityRefreshIntervalSeconds: 60}.Validate()) + // A negative activity-refresh interval is rejected. + s.Require().Error(Config{ActivityRefreshIntervalSeconds: -1}.Validate()) + // An activity-refresh interval equal to the idle window is rejected (must be strictly less). + s.Require().Error(Config{IdleTimeoutSeconds: 60, ActivityRefreshIntervalSeconds: 60}.Validate()) + // An activity-refresh interval exceeding the idle window is rejected. + s.Require().Error(Config{IdleTimeoutSeconds: 60, ActivityRefreshIntervalSeconds: 120}.Validate()) +} + +func (s *ConfigTestSuite) TestValidate_ActivityRefreshInvariantAcrossDefaults() { + // A zero means "use the default", so the invariant is checked against the resolved durations. + + // Gap A: a small configured idle with a defaulted (60s) refresh must be rejected, even though the + // refresh field itself is left unset. + s.Require().Error(Config{IdleTimeoutSeconds: 30}.Validate()) + // One second above the default refresh is accepted. + s.Require().NoError(Config{IdleTimeoutSeconds: 61}.Validate()) + + // Gap B: a defaulted (1800s) idle with a large configured refresh must be rejected, even though + // the idle field itself is left unset. + s.Require().Error(Config{ActivityRefreshIntervalSeconds: 3600}.Validate()) + // A large refresh below the defaulted idle window is accepted. + s.Require().NoError(Config{ActivityRefreshIntervalSeconds: 1799}.Validate()) +} + func (s *ConfigTestSuite) TestHandler_DecodeEmptyIsZero() { got, err := ConfigHandler{}.Decode(nil) s.Require().NoError(err) @@ -67,10 +102,11 @@ func (s *ConfigTestSuite) TestHandler_ValidateRejectsIncoherent() { } func (s *ConfigTestSuite) TestHandler_MergeWritableWins() { - readOnly := Config{IdleTimeoutSeconds: 1800, AbsoluteTimeoutSeconds: 28800} - writable := Config{IdleTimeoutSeconds: 600} + readOnly := Config{IdleTimeoutSeconds: 1800, AbsoluteTimeoutSeconds: 28800, ActivityRefreshIntervalSeconds: 60} + writable := Config{IdleTimeoutSeconds: 600, ActivityRefreshIntervalSeconds: 30} merged := ConfigHandler{}.Merge(readOnly, writable).(Config) // A positive writable field overrides read-only; an unset writable field keeps read-only. s.Equal(int64(600), merged.IdleTimeoutSeconds) s.Equal(int64(28800), merged.AbsoluteTimeoutSeconds) + s.Equal(int64(30), merged.ActivityRefreshIntervalSeconds) } diff --git a/backend/internal/flow/session/init.go b/backend/internal/flow/session/init.go index de4d6b5336..e281851a97 100644 --- a/backend/internal/flow/session/init.go +++ b/backend/internal/flow/session/init.go @@ -42,6 +42,9 @@ func Initialize(dbProvider provider.DBProviderInterface, deploymentID string, if timeouts.Absolute <= 0 { timeouts.Absolute = def.Absolute } + if timeouts.ActivityRefresh <= 0 { + timeouts.ActivityRefresh = def.ActivityRefresh + } store := newStore(dbProvider, deploymentID) return &service{ diff --git a/backend/internal/flow/session/service.go b/backend/internal/flow/session/service.go index 4256649445..7079e9d92d 100644 --- a/backend/internal/flow/session/service.go +++ b/backend/internal/flow/session/service.go @@ -56,9 +56,10 @@ type Service interface { SaveCheckpoint(ctx context.Context, in SaveCheckpointInput) (SaveCheckpointResult, error) // LoadCheckpoint fetches the session referenced by handle and its checkpoint context, refreshes - // the session's last-active timestamp and idle deadline, and records the joining participant with - // the grant's token family id (all best-effort). It errors when the session or its checkpoint - // context no longer exists. + // the session's last-active timestamp and idle deadline (throttled: skipped when the last refresh + // is within the activity-refresh window), and records the joining participant with the grant's + // token family id (all best-effort). It errors when the session or its checkpoint context no + // longer exists. LoadCheckpoint(ctx context.Context, handle, checkpoint, appID, tokenFamilyID string) ( *Session, *SessionContext, error) @@ -216,11 +217,19 @@ func (s *service) LoadCheckpoint(ctx context.Context, handle, checkpoint, appID, // Refresh last-active and slide the idle deadline under the optimistic-lock guard — touches // SESSION only. The absolute deadline is left unchanged so it keeps capping total lifetime. A // conflict here is non-fatal: the session loaded successfully. + // + // Throttle the write: within ActivityRefresh of the last persisted activity refresh, skip it. This + // hot path fires on every session reuse, and an unthrottled UPDATE per reuse is the dominant write + // load (and, on Postgres, the main source of dead tuples) on the session table. The persisted idle + // deadline then lags real activity by at most ActivityRefresh; config validation keeps that below + // the idle window so an active session is never skipped past its idle deadline. now := time.Now().UTC() - sess.LastActiveAt = now - sess.IdleExpiresAt = now.Add(s.timeouts.Idle) - if updErr := s.store.Update(ctx, sess); updErr != nil { - s.logger.Warn(ctx, "Failed to refresh session last-active timestamp", log.Error(updErr)) + if now.Sub(sess.LastActiveAt) >= s.timeouts.ActivityRefresh { + sess.LastActiveAt = now + sess.IdleExpiresAt = now.Add(s.timeouts.Idle) + if updErr := s.store.Update(ctx, sess); updErr != nil { + s.logger.Warn(ctx, "Failed to refresh session last-active timestamp", log.Error(updErr)) + } } // Record the joining application as a participant. When this reused session issues a token family, @@ -228,7 +237,8 @@ func (s *service) LoadCheckpoint(ctx context.Context, handle, checkpoint, appID, // these rows, so a token stamped with a tfid that has no persisted mapping would be unrevocable. // Fail closed in that case so the reuse does not issue an unrevocable family (the caller aborts the // load before publishing the tfid, forcing full re-authentication). Without a tfid there is nothing - // to revoke, so the write stays best-effort. + // to revoke, so the write stays best-effort. Either way it is not throttled with the activity + // refresh above: the upsert also registers an application joining the session for the first time. if partErr := s.recordParticipant(ctx, sess.SessionID, appID, tokenFamilyID, now); partErr != nil { if tokenFamilyID != "" { return nil, nil, fmt.Errorf("failed to record SSO session participant for token family: %w", partErr) @@ -355,7 +365,7 @@ func (s *service) establishSession(ctx context.Context, in SaveCheckpointInput) AuthenticatedAt: now, CreatedAt: now, LastActiveAt: now, - // The idle deadline slides on each activity touch; the absolute deadline is fixed here and + // The idle deadline slides on each activity refresh; the absolute deadline is fixed here and // caps the session's total lifetime. The resolver rejects a session past either deadline. IdleExpiresAt: now.Add(s.timeouts.Idle), AbsoluteExpiresAt: now.Add(s.timeouts.Absolute), diff --git a/backend/internal/flow/session/service_test.go b/backend/internal/flow/session/service_test.go index 44584c52b5..6dbcd80013 100644 --- a/backend/internal/flow/session/service_test.go +++ b/backend/internal/flow/session/service_test.go @@ -294,7 +294,7 @@ func (suite *ServiceTestSuite) TestLoadCheckpoint_Success() { suite.Require().NotNil(sess) suite.Require().NotNil(sc) - // Activity touch: last-active refreshed, idle slid forward, absolute unchanged. + // Activity refresh: last-active refreshed, idle slid forward, absolute unchanged. suite.Require().NotNil(updated) suite.False(updated.LastActiveAt.IsZero()) suite.True(updated.IdleExpiresAt.After(originalIdle)) @@ -367,6 +367,51 @@ func (suite *ServiceTestSuite) TestLoadCheckpoint_ParticipantErrorWithoutTokenFa suite.NotNil(sc) } +func (suite *ServiceTestSuite) TestLoadCheckpoint_ThrottledRefreshSkipsUpdate() { + svc, m := suite.newService() + // The last persisted activity refresh is within the activity-refresh window, so the idle-slide + // UPDATE is skipped. No Update expectation is set: the store mock fails the test if the throttled + // path writes. + recent := time.Now().UTC() + m.store.EXPECT().GetByHandle(mock.Anything, "handle-abc").Return(&Session{ + SessionID: "sess-1", HandleID: "handle-abc", State: StateActive, LastActiveAt: recent, + }, nil) + m.store.EXPECT().GetByCheckpoint(mock.Anything, "sess-1", "session"). + Return(&SessionContext{SessionID: "sess-1"}, nil) + m.store.EXPECT().Record(mock.Anything, mock.Anything).Return(nil) + + sess, sc, err := svc.LoadCheckpoint(context.Background(), "handle-abc", "session", "app-456", "") + suite.Require().NoError(err) + + suite.Require().NotNil(sess) + suite.Require().NotNil(sc) + suite.Equal(recent, sess.LastActiveAt, "a throttled load must leave the persisted last-active untouched") +} + +func (suite *ServiceTestSuite) TestLoadCheckpoint_WritesAfterRefreshWindow() { + svc, m := suite.newService() + // The last persisted activity refresh is older than the activity-refresh window, so the activity refresh persists. + stale := time.Now().UTC().Add(-2 * defaultActivityRefreshInterval) + originalIdle := stale.Add(time.Minute) + m.store.EXPECT().GetByHandle(mock.Anything, "handle-abc").Return(&Session{ + SessionID: "sess-1", HandleID: "handle-abc", State: StateActive, + LastActiveAt: stale, IdleExpiresAt: originalIdle, + }, nil) + m.store.EXPECT().GetByCheckpoint(mock.Anything, "sess-1", "session"). + Return(&SessionContext{SessionID: "sess-1"}, nil) + var updated *Session + m.store.EXPECT().Update(mock.Anything, mock.Anything).RunAndReturn( + func(_ context.Context, s *Session) error { updated = s; return nil }) + m.store.EXPECT().Record(mock.Anything, mock.Anything).Return(nil) + + _, _, err := svc.LoadCheckpoint(context.Background(), "handle-abc", "session", "app-456", "") + suite.Require().NoError(err) + + suite.Require().NotNil(updated, "an activity refresh past the throttle window must persist") + suite.True(updated.LastActiveAt.After(stale), "last-active slides forward") + suite.True(updated.IdleExpiresAt.After(originalIdle), "idle deadline slides forward") +} + // --- Terminate --- func (suite *ServiceTestSuite) TestTerminate_DeletesSessionAndPurges() { diff --git a/backend/internal/flow/session/state.go b/backend/internal/flow/session/state.go index 228fbd63d5..fa35f4ab6c 100644 --- a/backend/internal/flow/session/state.go +++ b/backend/internal/flow/session/state.go @@ -19,38 +19,66 @@ package session import ( + "math" "time" ) +// maxTimeoutSeconds is the largest configured second value that converts to a time.Duration without +// overflowing its int64 nanosecond representation. A larger value would wrap to a negative duration, +// so Config.Validate rejects it before it reaches NewTimeouts. +const maxTimeoutSeconds = math.MaxInt64 / int64(time.Second) + // DefaultIdleTimeout is the maximum inactivity period before a session expires. The idle deadline -// slides forward on each activity touch. +// slides forward on each activity refresh. const DefaultIdleTimeout = 30 * time.Minute // DefaultAbsoluteTimeout is the maximum lifetime of a session regardless of activity. It is fixed // at creation and never extended. It also bounds the transport cookie's max-age. const DefaultAbsoluteTimeout = 8 * time.Hour +// defaultActivityRefreshInterval is the built-in minimum time between persisted activity refreshes, +// used when the deployment does not configure session.activityRefreshIntervalSeconds. Within this +// window after the last persisted activity refresh, the activity refresh (the idle-slide UPDATE) is +// skipped to cut write amplification on the hot session-reuse path. +const defaultActivityRefreshInterval = 60 * time.Second + // Timeouts holds the resolved session lifetime durations used when minting and refreshing sessions. type Timeouts struct { - // Idle is the maximum inactivity period; the idle deadline slides on each activity touch. + // Idle is the maximum inactivity period; the idle deadline slides on each activity refresh. Idle time.Duration // Absolute is the maximum lifetime of a session regardless of activity. Absolute time.Duration + // ActivityRefresh is the minimum spacing between persisted activity refreshes; refreshes within + // this window of the last persisted one are skipped. Config validation keeps it below the idle + // window so an active session cannot be skipped past its idle deadline. + ActivityRefresh time.Duration +} + +// activityRefreshInterval resolves the activity-refresh spacing: the configured value (in seconds) +// when positive, otherwise the built-in default. The value is honored as configured; config +// validation is responsible for keeping it below the idle window. +func activityRefreshInterval(configuredSeconds int64) time.Duration { + if configuredSeconds > 0 { + return time.Duration(configuredSeconds) * time.Second + } + return defaultActivityRefreshInterval } // DefaultTimeouts returns the built-in default session timeouts. func DefaultTimeouts() Timeouts { return Timeouts{ - Idle: DefaultIdleTimeout, - Absolute: DefaultAbsoluteTimeout, + Idle: DefaultIdleTimeout, + Absolute: DefaultAbsoluteTimeout, + ActivityRefresh: activityRefreshInterval(0), } } // NewTimeouts builds session timeouts from per-field second values, falling back to the built-in // default for any non-positive value. The idle window is clamped to the absolute lifetime so the // pair is always valid — a defaulted idle can otherwise outrun a small configured absolute, which -// SessionConfig validation does not catch (it only compares the raw, positive config values). -func NewTimeouts(idleSeconds, absoluteSeconds int64) Timeouts { +// SessionConfig validation does not catch (it only compares the raw, positive config values). The +// activity-refresh interval uses activityRefreshSeconds when positive, else the built-in default. +func NewTimeouts(idleSeconds, absoluteSeconds, activityRefreshSeconds int64) Timeouts { t := DefaultTimeouts() if idleSeconds > 0 { t.Idle = time.Duration(idleSeconds) * time.Second @@ -61,5 +89,6 @@ func NewTimeouts(idleSeconds, absoluteSeconds int64) Timeouts { if t.Idle > t.Absolute { t.Idle = t.Absolute } + t.ActivityRefresh = activityRefreshInterval(activityRefreshSeconds) return t } diff --git a/backend/internal/flow/session/state_test.go b/backend/internal/flow/session/state_test.go index 2a5223b1e7..3dd9e4a155 100644 --- a/backend/internal/flow/session/state_test.go +++ b/backend/internal/flow/session/state_test.go @@ -38,34 +38,46 @@ func TestStateTestSuite(t *testing.T) { func (s *StateTestSuite) TestNewTimeouts_Defaults() { // Non-positive values fall back to the built-in defaults. - got := NewTimeouts(-5, 0) + got := NewTimeouts(-5, 0, 0) s.Equal(DefaultTimeouts(), got) } func (s *StateTestSuite) TestNewTimeouts_Overrides() { - got := NewTimeouts(60, 600) + got := NewTimeouts(60, 600, 0) s.Equal(60*time.Second, got.Idle) s.Equal(600*time.Second, got.Absolute) } func (s *StateTestSuite) TestNewTimeouts_PartialOverride() { - got := NewTimeouts(60, 0) + got := NewTimeouts(60, 0, 0) s.Equal(60*time.Second, got.Idle) s.Equal(DefaultAbsoluteTimeout, got.Absolute, "unset absolute falls back to default") } func (s *StateTestSuite) TestNewTimeouts_IdleClampedToAbsolute() { // A small configured absolute with a defaulted (larger) idle must not yield idle > absolute. - got := NewTimeouts(0, 60) + got := NewTimeouts(0, 60, 0) s.Equal(60*time.Second, got.Absolute) s.Equal(60*time.Second, got.Idle, "idle is clamped to the absolute lifetime") // An explicit idle larger than absolute is likewise clamped. - got = NewTimeouts(600, 300) + got = NewTimeouts(600, 300, 0) s.Equal(300*time.Second, got.Idle) s.Equal(300*time.Second, got.Absolute) } +func (s *StateTestSuite) TestNewTimeouts_ActivityRefreshInterval() { + // An unset interval yields the built-in default. + s.Equal(defaultActivityRefreshInterval, NewTimeouts(0, 0, 0).ActivityRefresh) + + // A configured interval is honored exactly. + s.Equal(20*time.Second, NewTimeouts(1800, 28800, 20).ActivityRefresh) + + // A configured interval above half the idle window is honored as-is, not clamped down. + s.Equal(1200*time.Second, NewTimeouts(1800, 28800, 1200).ActivityRefresh, + "a configured interval is honored, not clamped") +} + func (s *StateTestSuite) TestSessionContext_PayloadRoundTrip() { c := SessionContext{ SessionID: "sess-1",