diff --git a/internal/api/handler_history.go b/internal/api/handler_history.go index 3e9079516..868ac4a55 100644 --- a/internal/api/handler_history.go +++ b/internal/api/handler_history.go @@ -10,6 +10,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/auth" "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/internal/runtime" "github.com/LeanerCloud/CUDly/pkg/logging" "github.com/aws/aws-lambda-go/events" ) @@ -55,12 +56,14 @@ func (h *Handler) getHistory(ctx context.Context, req *events.LambdaFunctionURLR // scheduled_date) so the two halves of the merged response are // consistently scoped (issue #701). // - // Stale pending/notified executions are expired asynchronously AFTER the - // response is assembled so the GET handler is a pure read (issue #1032): - // the caller sees current DB state; the transitions fire in a goroutine - // and the next History load reflects the updated status. + // Stale pending/notified executions are expired AFTER the response is + // assembled so the GET handler is a pure read (issue #1032): the caller + // sees current DB state and the next History load reflects the updated + // status. On servers the transitions fire in a goroutine; on Lambda they + // run synchronously before returning because the execution environment + // freezes once the response is out (issue #1170). extra, staleExecs := h.fetchExecutionsAsHistory(ctx, filters) - h.expireStaleExecutionsAsync(staleExecs) + h.expireStaleExecutions(staleExecs) all := make([]config.PurchaseHistoryRecord, 0, len(completed)+len(extra)) all = append(all, completed...) @@ -132,11 +135,11 @@ const approvalExpiryWindow = 7 * 24 * time.Hour // to "multiple" because a single execution can span providers. The approver // address is looked up from global config once and attached to pending/ // notified rows so the UI can tell the user exactly whose inbox holds the -// approval link. Stale pending/notified executions are expired asynchronously -// after this call returns (see expireStaleExecutionsAsync) so the GET is a -// pure read: the response rows carry the pre-transition status and the -// next request sees the updated state. A listing error is logged and -// skipped — completed history must still render. +// approval link. Stale pending/notified executions are expired after this +// call returns (see expireStaleExecutions) so the GET is a pure read: +// the response rows carry the pre-transition status and the next request +// sees the updated state. A listing error is logged and skipped — completed +// history must still render. // // The filter set (issue #701) is applied in Go against the synthetic row: // provider via the recs' collapsed provider, account via CloudAccountID, @@ -164,10 +167,10 @@ func (h *Handler) fetchExecutionsAsHistory(ctx context.Context, filters historyF if exec.Status == "completed" && exec.Error == "" { continue } - // Collect stale pending/notified executions for the post-response - // async expire sweep. We do NOT mutate status here to keep the GET - // read-only: the response reflects current DB state; the background - // goroutine fires the transition so the next request sees "expired". + // Collect stale pending/notified executions for the post-assembly + // expire sweep. We do NOT mutate status here to keep the GET + // read-only: the response reflects current DB state; the sweep + // fires the transition so the next request sees "expired". if isStaleExecution(exec) { staleExecs = append(staleExecs, exec) } @@ -185,8 +188,8 @@ func (h *Handler) fetchExecutionsAsHistory(ctx context.Context, filters historyF // isStaleExecution reports whether the execution is a pending/notified // approval older than approvalExpiryWindow that should be transitioned to -// "expired". Extracted so both fetchExecutionsAsHistory and the async sweep -// share one staleness check. +// "expired". Extracted so both fetchExecutionsAsHistory and the expire +// sweep (sync on Lambda, async on servers) share one staleness check. func isStaleExecution(exec config.PurchaseExecution) bool { if exec.Status != "pending" && exec.Status != "notified" { return false @@ -194,33 +197,51 @@ func isStaleExecution(exec config.PurchaseExecution) bool { return time.Since(exec.ScheduledDate) >= approvalExpiryWindow } -// expireStaleExecutionsAsync fires TransitionExecutionStatus for each stale -// execution in a best-effort goroutine that outlives the request context. -// Using context.Background() ensures the transitions are not canceled when -// the HTTP handler returns. Errors are logged and skipped — a missed -// transition leaves the row "pending" until the next History load, which is -// better than blocking the read response. +// expireStaleExecutions fires TransitionExecutionStatus for each stale +// execution. On long-running servers this happens in a best-effort goroutine +// that outlives the request context: context.Background() ensures the +// transitions are not canceled when the HTTP handler returns. On Lambda that +// guarantee does not hold: the execution environment freezes as soon as the +// response is returned, so a background goroutine would be suspended +// mid-sweep and stale rows could stay "pending" indefinitely (issue #1170). +// There the sweep runs synchronously before the handler returns, mirroring +// the SWR cache's isLambda gate (ri_utilization_cache.go) and using the same +// runtime.IsLambda detection helper. The sweep is a handful of cheap UPDATEs, +// so the synchronous cost on Lambda is negligible. Errors are logged and +// skipped — a missed transition leaves the row "pending" until the next +// History load, which is better than failing the read response. // -// The goroutine is idempotent per execution ID: TransitionExecutionStatus is +// The sweep is idempotent per execution ID: TransitionExecutionStatus is // guarded by the FROM-status list ("pending","notified"), so a concurrent // caller that wins the race causes the loser's update to affect 0 rows and // return an error, which is already handled by the Warnf below. Two -// simultaneous GET requests can both spawn a goroutine for the same stale -// row; only one transition commits — this is safe and expected. -func (h *Handler) expireStaleExecutionsAsync(staleExecs []config.PurchaseExecution) { +// simultaneous GET requests can both sweep the same stale row; only one +// transition commits — this is safe and expected. +func (h *Handler) expireStaleExecutions(staleExecs []config.PurchaseExecution) { if len(staleExecs) == 0 { return } - go func() { - ctx := context.Background() - for _rvc := range staleExecs { - exec := staleExecs[_rvc] - _, err := h.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"pending", "notified"}, "expired", nil) - if err != nil { - logging.Warnf("history: async expire of execution %s failed: %v", exec.ExecutionID, err) - } + if runtime.IsLambda() { + h.expireStaleExecutionsSweep(staleExecs) + return + } + go h.expireStaleExecutionsSweep(staleExecs) +} + +// expireStaleExecutionsSweep is the shared sweep body for both branches of +// expireStaleExecutions. It deliberately uses context.Background() +// rather than the request context: on servers the goroutine outlives the +// request, and on Lambda the request context may carry a deadline that +// should not abort the best-effort transitions. +func (h *Handler) expireStaleExecutionsSweep(staleExecs []config.PurchaseExecution) { + ctx := context.Background() + for _rvc := range staleExecs { + exec := staleExecs[_rvc] + _, err := h.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"pending", "notified"}, "expired", nil) + if err != nil { + logging.Warnf("history: expire sweep for execution %s failed: %v", exec.ExecutionID, err) } - }() + } } // resolveUserEmails builds a map of user-ID to email by calling GetUser once diff --git a/internal/api/handler_history_test.go b/internal/api/handler_history_test.go index c77369afe..ed5721c4c 100644 --- a/internal/api/handler_history_test.go +++ b/internal/api/handler_history_test.go @@ -1838,3 +1838,125 @@ func TestSummarizePurchaseHistory_CancelPendingDoesNotChangeKPIs(t *testing.T) { assert.Equal(t, before.TotalCompleted, after.TotalCompleted, "canceling a pending purchase must not change TotalCompleted (issue #736)") } + +// TestHandler_getHistory_ExpireIfStale_LambdaGuard covers the Lambda guard on +// the stale-approval expiry sweep (issue #1170, COR-06). On Lambda the +// execution environment freezes as soon as the response is returned, so the +// sweep must complete synchronously before getHistory returns; on long-running +// servers it must stay asynchronous so the read response is never blocked on +// the transitions. Detection reuses runtime.IsLambda (AWS_LAMBDA_RUNTIME_API), +// the same helper the SWR cache goroutine gate uses, so the sub-tests flip +// that env var via t.Setenv. +func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) { + approverEmail := "ops@example.com" + staleID := "stale-exec-lambda-guard" + staleExec := func() config.PurchaseExecution { + return config.PurchaseExecution{ + ExecutionID: staleID, + Status: "pending", + ScheduledDate: time.Now().Add(-8 * 24 * time.Hour), + TotalUpfrontCost: 200.0, + EstimatedSavings: 20.0, + Recommendations: []config.RecommendationRecord{ + {Provider: "aws", Service: "rds", Region: "us-east-1"}, + }, + } + } + + t.Run("Lambda: sweep completes synchronously before the handler returns", func(t *testing.T) { + // Any non-empty value marks the process as running on Lambda + // (runtime.IsLambda checks presence, not content). + t.Setenv("AWS_LAMBDA_RUNTIME_API", "127.0.0.1:9001") + + ctx := context.Background() + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + expired := staleExec() + expired.Status = "expired" + + mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). + Return([]config.PurchaseExecution{staleExec()}, nil) + mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) + mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired", + (*string)(nil), + ).Return(&expired, nil).Once() + + mockAuth, req := adminHistoryReq(ctx) + handler := &Handler{auth: mockAuth, config: mockStore} + + result, err := handler.getHistory(ctx, req, map[string]string{}) + require.NoError(t, err) + + // No channel wait: on Lambda the transition must already have fired + // by the time getHistory returns. Pre-fix this fails because the + // sweep ran in a goroutine the frozen sandbox never resumes. + mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1) + + historyResp := result.(HistoryResponse) + require.Len(t, historyResp.Purchases, 1) + assert.Equal(t, "pending", historyResp.Purchases[0].Status, + "GET stays a pure read on Lambda too: response carries the pre-transition status") + }) + + t.Run("non-Lambda: sweep stays asynchronous and never blocks the response", func(t *testing.T) { + // Explicitly clear the marker so the test is deterministic even if + // the outer environment sets it; t.Setenv restores it afterwards. + t.Setenv("AWS_LAMBDA_RUNTIME_API", "") + + ctx := context.Background() + mockStore := new(MockConfigStore) + expired := staleExec() + expired.Status = "expired" + + // The transition blocks until released. If the sweep ran + // synchronously off-Lambda (a regression of issue #1032's pure-read + // guarantee), getHistory would block on it and the watchdog below + // would fire. + release := make(chan struct{}) + swept := make(chan struct{}) + mockStore.On("GetAllPurchaseHistory", ctx, 100).Return([]config.PurchaseHistoryRecord{}, nil) + mockStore.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything). + Return([]config.PurchaseExecution{staleExec()}, nil) + mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: &approverEmail}, nil) + mockStore.On("TransitionExecutionStatus", mock.Anything, staleID, []string{"pending", "notified"}, "expired", + (*string)(nil), + ).Run(func(_ mock.Arguments) { + <-release + close(swept) + }).Return(&expired, nil).Once() + + mockAuth, req := adminHistoryReq(ctx) + handler := &Handler{auth: mockAuth, config: mockStore} + + done := make(chan struct{}) + var result any + var err error + go func() { + result, err = handler.getHistory(ctx, req, map[string]string{}) + close(done) + }() + + select { + case <-done: + // getHistory returned while the transition was still blocked: + // the sweep is asynchronous, as required off-Lambda. + case <-time.After(5 * time.Second): + t.Fatal("getHistory blocked on the expiry sweep off-Lambda - sweep must stay asynchronous") + } + + close(release) + select { + case <-swept: + case <-time.After(5 * time.Second): + t.Fatal("background expiry sweep did not complete within 5s") + } + + require.NoError(t, err) + mockStore.AssertNumberOfCalls(t, "TransitionExecutionStatus", 1) + mockStore.AssertExpectations(t) + historyResp := result.(HistoryResponse) + require.Len(t, historyResp.Purchases, 1) + assert.Equal(t, "pending", historyResp.Purchases[0].Status) + }) +}