Forward the SSO-Check node's session and checkpoint reads to the Session node

backend/internal/flow/common/constants.go

@@ -309,6 +309,12 @@ const (
 	// ForwardedDataKeyExpiryMinutes is the key for the OTP expiry duration (in minutes) inside the
 	// ForwardedData[ForwardedDataKeyTemplateData] map forwarded by OTPExecutor to sender executors.
 	ForwardedDataKeyExpiryMinutes = "expiryMinutes"
+	// ForwardedDataKeySSOSession holds the SSO session the SSO-Check node resolved, forwarded to the
+	// paired Session node so it restores the checkpoint without reading the same row again.
+	ForwardedDataKeySSOSession = "ssoSession"
+	// ForwardedDataKeySSOSessionContext holds the checkpoint context the SSO-Check node fetched,
+	// forwarded alongside ForwardedDataKeySSOSession.
+	ForwardedDataKeySSOSessionContext = "ssoSessionContext"
 )
 
 // InterceptorStatus represents the outcome of an interceptor execution.

backend/internal/flow/executor/session_executor.go

@@ -224,6 +224,18 @@ func setHandleOut(execResp *providers.ExecutorResponse, handle string) {
 	execResp.EngineData[common.RuntimeKeySSOSessionHandle] = handle
 }
 
+// readForwardedSSOData returns the session and checkpoint context the paired SSO-Check node read and
+// forwarded, or nils when this node was not reached directly from it. The service re-reads whatever it
+// is not given, so a missing or partial handover costs a query rather than correctness.
+func readForwardedSSOData(ctx *providers.NodeContext) (*session.Session, *session.SessionContext) {
+	if ctx.ForwardedData == nil {
+		return nil, nil
+	}
+	forwardedSession, _ := ctx.ForwardedData[common.ForwardedDataKeySSOSession].(*session.Session)
+	forwardedContext, _ := ctx.ForwardedData[common.ForwardedDataKeySSOSessionContext].(*session.SessionContext)
+	return forwardedSession, forwardedContext
+}
+
 // loadCheckpoint loads a checkpoint's saved flow state into the execution context so downstream
 // nodes continue with the authenticated subject and claims. The SSO session service fetches the
 // session and its checkpoint context (and refreshes the session's activity); this executor
@@ -234,28 +246,37 @@ func (e *sessionExecutor) loadCheckpoint(ctx *providers.NodeContext, execResp *p
 	// An SSO reuse still issues a fresh grant, so mint a new token family id and record it against the
 	// joining participant. It is published onto RuntimeData after the snapshot replay below.
 	tokenFamilyID := e.resolveTokenFamilyID(ctx, logger)
-	sess, sc, err := e.sso.LoadCheckpoint(ctx.Context, handle, checkpoint, ctx.Application.ID, tokenFamilyID)
+	forwardedSession, forwardedContext := readForwardedSSOData(ctx)
+	ssoSession, snapshot, err := e.sso.LoadCheckpoint(ctx.Context, session.LoadCheckpointInput{
+		Handle:        handle,
+		Checkpoint:    checkpoint,
+		AppID:         ctx.Application.ID,
+		TokenFamilyID: tokenFamilyID,
+		Session:       forwardedSession,
+		Context:       forwardedContext,
+	})
 	if err != nil {
 		return err
 	}
 
 	// Rehydrate the AuthUser from the snapshot verbatim — it was stored as-is — so downstream nodes
 	// continue with the same subject and attributes this session resolved when the checkpoint was saved.
 	var authUser providers.AuthUser
-	if err := json.Unmarshal(sc.AuthUser, &authUser); err != nil {
+	if err := json.Unmarshal(snapshot.AuthUser, &authUser); err != nil {
 		return fmt.Errorf("failed to rehydrate subject reference from snapshot: %w", err)
 	}
 	execResp.AuthUser = authUser
 
 	// Replay the snapshotted RuntimeData (the effective attribute set captured at save) so downstream
 	// nodes see the same attributes the fresh path produced.
-	for k, v := range sc.RuntimeData {
+	for k, v := range snapshot.RuntimeData {
 		execResp.RuntimeData[k] = v
 	}
 	// auth_time comes from the lean session, not the context. Set it after the RuntimeData replay so
 	// the live, session-derived value wins over any stale snapshot copy.
-	if !sess.AuthenticatedAt.IsZero() {
-		execResp.RuntimeData[common.RuntimeKeyAuthTime] = strconv.FormatInt(sess.AuthenticatedAt.Unix(), 10)
+	if !ssoSession.AuthenticatedAt.IsZero() {
+		execResp.RuntimeData[common.RuntimeKeyAuthTime] =
+			strconv.FormatInt(ssoSession.AuthenticatedAt.Unix(), 10)
 	}
 	// Publish the freshly minted token family id after the snapshot replay so it is never shadowed by
 	// a stale copy (the tfid is excluded from the snapshot, so this is the only source).

backend/internal/flow/executor/session_executor_test.go

@@ -331,7 +331,9 @@ func (suite *SessionExecutorTestSuite) TestSSOLoad() {
 	snapAuthUser := `{"default":{"entityReference":{"entityId":"user-2","ouId":"ou-9","type":"person"},` +
 		`"attributes":{"attributes":{"email":{"value":"bob@example.com"}}}}}`
 	sso := sessionmock.NewServiceMock(suite.T())
-	sso.EXPECT().LoadCheckpoint(mock.Anything, "handle-abc", "session", "app-456", mock.Anything).Return(
+	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.MatchedBy(func(in session.LoadCheckpointInput) bool {
+		return in.Handle == "handle-abc" && in.Checkpoint == "session" && in.AppID == "app-456"
+	})).Return(
 		&session.Session{
 			SessionID: "sess-1", SubjectID: "user-2", HandleID: "handle-abc",
 			AuthenticatedAt: time.Unix(1700000000, 0).UTC(),
@@ -368,11 +370,56 @@ func (suite *SessionExecutorTestSuite) TestSSOLoad() {
 	suite.Equal("1700000000", resp.RuntimeData[common.RuntimeKeyAuthTime])
 }
 
+// TestSSOLoad_PassesForwardedReadsToService is the executor half of the reuse-path read reduction: the
+// rows the SSO-Check node put on ForwardedData must reach the service, which then skips both reads.
+func (suite *SessionExecutorTestSuite) TestSSOLoad_PassesForwardedReadsToService() {
+	forwardedSession := &session.Session{SessionID: "sess-1", HandleID: "handle-abc"}
+	forwardedContext := &session.SessionContext{
+		SessionID: "sess-1", CheckpointID: "session", AuthUser: json.RawMessage("{}"),
+	}
+
+	sso := sessionmock.NewServiceMock(suite.T())
+	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.MatchedBy(func(in session.LoadCheckpointInput) bool {
+		return in.Session == forwardedSession && in.Context == forwardedContext
+	})).Return(forwardedSession, forwardedContext, nil)
+	exec := suite.newExecutor(sso, managermock.NewAuthnProviderManagerMock(suite.T()))
+
+	ctx := ssoLoadCtx()
+	ctx.ForwardedData = map[string]interface{}{
+		common.ForwardedDataKeySSOSession:        forwardedSession,
+		common.ForwardedDataKeySSOSessionContext: forwardedContext,
+	}
+
+	_, err := exec.Execute(ctx)
+
+	suite.Require().NoError(err)
+}
+
+// TestSSOLoad_ForwardsNothingWhenAbsent covers a Session node reached without the paired SSO-Check
+// node's handover: the service is asked to read both rows itself rather than being handed junk.
+func (suite *SessionExecutorTestSuite) TestSSOLoad_ForwardsNothingWhenAbsent() {
+	sso := sessionmock.NewServiceMock(suite.T())
+	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.MatchedBy(func(in session.LoadCheckpointInput) bool {
+		return in.Session == nil && in.Context == nil
+	})).Return(
+		&session.Session{SessionID: "sess-1", HandleID: "handle-abc"},
+		&session.SessionContext{SessionID: "sess-1", CheckpointID: "session", AuthUser: json.RawMessage("{}")},
+		nil)
+	exec := suite.newExecutor(sso, managermock.NewAuthnProviderManagerMock(suite.T()))
+
+	ctx := ssoLoadCtx()
+	ctx.ForwardedData = map[string]interface{}{"unrelated": 42}
+
+	_, err := exec.Execute(ctx)
+
+	suite.Require().NoError(err)
+}
+
 // TestSSOLoad_ErrorFailsFlow covers a load failure surfacing as a server error so the task-execution
 // node fails the flow (the credential steps were already skipped).
 func (suite *SessionExecutorTestSuite) TestSSOLoad_ErrorFailsFlow() {
 	sso := sessionmock.NewServiceMock(suite.T())
-	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.Anything).
 		Return(nil, nil, errors.New("resolved session no longer exists"))
 	exec := suite.newExecutor(sso, managermock.NewAuthnProviderManagerMock(suite.T()))
 
@@ -386,7 +433,7 @@ func (suite *SessionExecutorTestSuite) TestSSOLoad_ErrorFailsFlow() {
 // reconstruct the subject, so the flow fails.
 func (suite *SessionExecutorTestSuite) TestSSOLoad_RehydrateErrorFailsFlow() {
 	sso := sessionmock.NewServiceMock(suite.T())
-	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(
+	sso.EXPECT().LoadCheckpoint(mock.Anything, mock.Anything).Return(
 		&session.Session{SessionID: "sess-1", HandleID: "handle-abc"},
 		&session.SessionContext{SessionID: "sess-1", AuthUser: json.RawMessage("not-json")}, nil)
 	exec := suite.newExecutor(sso, managermock.NewAuthnProviderManagerMock(suite.T()))

backend/internal/flow/executor/sso_check_executor.go

@@ -89,16 +89,24 @@ func (e *ssoCheckExecutor) Execute(ctx *providers.NodeContext) (*providers.Execu
 		execResp.RuntimeData[common.RuntimeKeySSOSessionHandle] = resolved.HandleID
 	}
 
-	present := false
+	var snapshot *session.SessionContext
 	if resolved != nil && checkpoint != "" {
-		if present, err = e.sso.HasCheckpoint(ctx.Context, resolved.SessionID, checkpoint); err != nil {
+		if snapshot, err = e.sso.FindCheckpoint(ctx.Context, resolved.SessionID, checkpoint); err != nil {
 			return execResp, err
 		}
 	}
 
-	if present {
+	if snapshot != nil {
 		execResp.Status = providers.ExecComplete
 		execResp.RuntimeData[presentKey] = dataValueTrue
+		// Hand the two rows this node just read to the paired Session node, which needs the same ones
+		// to restore the checkpoint. ForwardedData reaches the immediate next node only, so this is
+		// set on the Skip outcome alone: on the Authenticate outcome the flow prompts and suspends,
+		// and forwarded data that outlived the node would be persisted with the flow context.
+		execResp.ForwardedData = map[string]interface{}{
+			common.ForwardedDataKeySSOSession:        resolved,
+			common.ForwardedDataKeySSOSessionContext: snapshot,
+		}
 		logger.Debug(ctx.Context, "Live SSO checkpoint present; routing to the Skip outcome",
 			log.String("flowId", in.FlowID),
 			log.String("checkpoint", checkpoint))

backend/internal/flow/executor/sso_check_executor_test.go

@@ -87,14 +87,18 @@ func (suite *SSOCheckExecutorTestSuite) assertAbsent(resp *providers.ExecutorRes
 	suite.Equal("false",
 		resp.RuntimeData[common.SSOCheckpointKey(common.RuntimeKeySSOSessionPresent, "session")])
 	suite.Empty(resp.RuntimeData[common.RuntimeKeySSOSessionHandle])
+	// The Authenticate outcome prompts and suspends the flow, and forwarded data that outlived the node
+	// would be persisted with the flow context, so nothing is forwarded on this path.
+	suite.Empty(resp.ForwardedData)
 }
 
 // TestPresent covers a live session that already holds this checkpoint: routes to Skip and shares the
 // handle so the paired Session node loads the saved state.
 func (suite *SSOCheckExecutorTestSuite) TestPresent() {
 	sso := sessionmock.NewServiceMock(suite.T())
 	sso.EXPECT().Resolve(mock.Anything, "handle-abc", "flow-1", 3, mock.Anything).Return(liveSession(), nil)
-	sso.EXPECT().HasCheckpoint(mock.Anything, "sess-1", "session").Return(true, nil)
+	sso.EXPECT().FindCheckpoint(mock.Anything, "sess-1", "session").
+		Return(&session.SessionContext{SessionID: "sess-1", CheckpointID: "session"}, nil)
 	exec := suite.newExecutor(sso)
 
 	resp, err := exec.Execute(ssoNodeContext())
@@ -104,6 +108,14 @@ func (suite *SSOCheckExecutorTestSuite) TestPresent() {
 	suite.Equal(dataValueTrue,
 		resp.RuntimeData[common.SSOCheckpointKey(common.RuntimeKeySSOSessionPresent, "session")])
 	suite.Equal("handle-abc", resp.RuntimeData[common.RuntimeKeySSOSessionHandle])
+	// Both rows this node read are forwarded to the paired Session node so the load path does not read
+	// them again. Losing this makes the reuse correct but two queries more expensive.
+	forwardedSession, ok := resp.ForwardedData[common.ForwardedDataKeySSOSession].(*session.Session)
+	suite.Require().True(ok, "the resolved session must be forwarded to the Session node")
+	suite.Equal("sess-1", forwardedSession.SessionID)
+	forwardedContext, ok := resp.ForwardedData[common.ForwardedDataKeySSOSessionContext].(*session.SessionContext)
+	suite.Require().True(ok, "the checkpoint context must be forwarded to the Session node")
+	suite.Equal("session", forwardedContext.CheckpointID)
 }
 
 // TestAbsentCheckpointNotPresent covers a live session that lacks this checkpoint: the node routes to
@@ -113,7 +125,7 @@ func (suite *SSOCheckExecutorTestSuite) TestAbsentCheckpointNotPresent() {
 	sso := sessionmock.NewServiceMock(suite.T())
 	sso.EXPECT().Resolve(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(liveSession(), nil)
-	sso.EXPECT().HasCheckpoint(mock.Anything, "sess-1", "session").Return(false, nil)
+	sso.EXPECT().FindCheckpoint(mock.Anything, "sess-1", "session").Return(nil, nil)
 	exec := suite.newExecutor(sso)
 
 	resp, err := exec.Execute(ssoNodeContext())
@@ -170,14 +182,14 @@ func (suite *SSOCheckExecutorTestSuite) TestResolverErrorFailsFlow() {
 	suite.Contains(err.Error(), "store down")
 }
 
-// TestCheckpointListErrorFailsFlow covers a checkpoint-existence lookup failure on a live session:
+// TestCheckpointLookupErrorFailsFlow covers a checkpoint lookup failure on a live session:
 // Execute returns a Go error rather than skipping on incomplete information.
-func (suite *SSOCheckExecutorTestSuite) TestCheckpointListErrorFailsFlow() {
+func (suite *SSOCheckExecutorTestSuite) TestCheckpointLookupErrorFailsFlow() {
 	sso := sessionmock.NewServiceMock(suite.T())
 	sso.EXPECT().Resolve(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(liveSession(), nil)
-	sso.EXPECT().HasCheckpoint(mock.Anything, mock.Anything, mock.Anything).
-		Return(false, errors.New("store down"))
+	sso.EXPECT().FindCheckpoint(mock.Anything, mock.Anything, mock.Anything).
+		Return(nil, errors.New("store down"))
 	exec := suite.newExecutor(sso)
 
 	_, err := exec.Execute(ssoNodeContext())