diff --git a/internal/ctxutil/detached.go b/internal/ctxutil/detached.go new file mode 100644 index 000000000..67daf3662 --- /dev/null +++ b/internal/ctxutil/detached.go @@ -0,0 +1,19 @@ +// Package ctxutil holds small context helpers shared across bounded contexts. +// Keep it minimal: add a helper here only when more than one package needs it. +package ctxutil + +import ( + "context" + "time" +) + +// DetachedWithTimeout returns a context that keeps ctx's values but drops its +// cancellation and deadline, bounded instead by a fresh timeout of d. +// +// Use it for work that must outlive the request or caller that started it, +// such as a provider call that commits money and the terminal write recording +// its outcome: a client disconnect or a per-request timeout must not cancel +// either midway. Always defer the returned CancelFunc. +func DetachedWithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), d) +} diff --git a/internal/ctxutil/detached_test.go b/internal/ctxutil/detached_test.go new file mode 100644 index 000000000..ffeaacf9c --- /dev/null +++ b/internal/ctxutil/detached_test.go @@ -0,0 +1,37 @@ +package ctxutil + +import ( + "context" + "testing" + "time" +) + +type ctxKey struct{} + +func TestDetachedWithTimeout(t *testing.T) { + parent, cancelParent := context.WithTimeout(context.WithValue(context.Background(), ctxKey{}, "req-1"), time.Millisecond) + cancelParent() + + start := time.Now() + ctx, cancel := DetachedWithTimeout(parent, time.Minute) + defer cancel() + + if err := ctx.Err(); err != nil { + t.Fatalf("parent cancellation must not reach the detached context, got %v", err) + } + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("detached context must carry its own deadline") + } + if d := deadline.Sub(start); d < 59*time.Second || d > 61*time.Second { + t.Fatalf("deadline must be ~1m from now (not the parent's), got %s", d) + } + if got := ctx.Value(ctxKey{}); got != "req-1" { + t.Fatalf("detached context must keep parent values, got %v", got) + } + + cancel() + if ctx.Err() == nil { + t.Fatal("the returned CancelFunc must cancel the detached context") + } +} diff --git a/internal/server/app.go b/internal/server/app.go index c2ca3e804..d7eedbbcd 100644 --- a/internal/server/app.go +++ b/internal/server/app.go @@ -93,6 +93,10 @@ type Application struct { // Only tests set it. initRetryDelay time.Duration + // scheduledTasks tracks detached HTTP-triggered scheduled tasks so server + // shutdown can cancel and wait for them (#784). + scheduledTasks scheduledTaskTracker + // encKeySource is the env var name that resolved the credential encryption // key (e.g. "CREDENTIAL_ENCRYPTION_KEY_SECRET_NAME"). Set during // reinitializeAfterConnect; surfaced via /health. diff --git a/internal/server/http.go b/internal/server/http.go index 4678d90c6..95933b3e0 100644 --- a/internal/server/http.go +++ b/internal/server/http.go @@ -159,6 +159,14 @@ func (app *Application) initializeUntilReady(ctx context.Context, init func(cont // background initializer. A variable so tests can shorten it. var shutdownGracePeriod = 30 * time.Second +// scheduledTaskUnwindShare reserves 1/N of shutdownGracePeriod (5s of 30s) for +// detached scheduled tasks to unwind after they are canceled at the end of +// the drain window. The whole sequence stays inside shutdownGracePeriod, which +// matches Container Apps' default 30s termination grace. Cloud Run sends +// SIGKILL 10s after SIGTERM, so there a task still running at that point is +// hard-killed; the advisory lock and row CAS make the next tick safe. +const scheduledTaskUnwindShare = 6 + // StartHTTPServer starts the HTTP server with graceful shutdown on SIGINT/SIGTERM. // It blocks until the server exits cleanly. In container orchestrators (Cloud Run, // Container Apps, Fargate) SIGTERM is the normal stop signal; without this wiring @@ -192,9 +200,19 @@ func StartHTTPServer(app *Application, port int) error { log.Printf("Shutdown signal received; draining HTTP server (30s grace)...") shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGracePeriod) defer cancel() - if err := server.Shutdown(shutdownCtx); err != nil { + // Drain for most of the grace period, leaving the last part for + // canceled scheduled tasks to unwind. + drainCtx, cancelDrain := context.WithTimeout(shutdownCtx, shutdownGracePeriod-shutdownGracePeriod/scheduledTaskUnwindShare) + defer cancelDrain() + if err := server.Shutdown(drainCtx); err != nil { log.Printf("HTTP server forced shutdown: %v", err) } + // Scheduled tasks run detached from their request (#784), so draining + // does not stop them. Cancel any still running and wait for them to + // return before main closes the database under them. + if !app.scheduledTasks.stop(shutdownCtx) { + log.Printf("Scheduled tasks did not stop within the shutdown grace period") + } // The initializer shares app.DB with app.Close; let it exit first. select { case <-initDone: @@ -317,8 +335,6 @@ func (app *Application) handleScheduledHTTP(w http.ResponseWriter, r *http.Reque return } - ctx := r.Context() - // Extract task type from URL path: /api/scheduled/{task_type}. // TrimPrefix is cleaner than splitting and indexing parts[2]; it also // avoids the length-guard dance while remaining robust to extra slashes @@ -338,6 +354,16 @@ func (app *Application) handleScheduledHTTP(w http.ResponseWriter, r *http.Reque return } + // Run the task on a context detached from the request (#784). The + // scheduler gives up after its attempt deadline (Cloud Scheduler 320s, + // Logic Apps 120s) and drops the connection, which cancels r.Context(). + // Inheriting that cancellation could abort a task midway (e.g. between a + // provider call and the write recording it). The task stays bounded by + // scheduledTaskHTTPTimeout and is canceled on server shutdown. + ctx, done := app.scheduledTasks.start(r.Context(), scheduledTaskHTTPTimeout) + defer done() + extendScheduledWriteDeadline(w) + // Ensure database connection is established (lazy initialization) if err = app.ensureDB(ctx); err != nil { log.Printf("Failed to establish database connection: %v", err) diff --git a/internal/server/http_scheduled_context_test.go b/internal/server/http_scheduled_context_test.go new file mode 100644 index 000000000..044bd9932 --- /dev/null +++ b/internal/server/http_scheduled_context_test.go @@ -0,0 +1,234 @@ +package server + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "regexp" + "strconv" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/purchase" + "github.com/LeanerCloud/cloud-commitments-platform/internal/testutil" +) + +// TestHandleScheduledHTTP_RequestCancelDoesNotCancelTask is the regression +// test for issue #784: when the scheduler (Cloud Scheduler, Azure Logic +// Apps) gives up and drops the connection, the request context is canceled. +// The scheduled task, which may be mid-purchase, must keep running on a +// context that does not inherit that cancellation, but it must stay bounded. +func TestHandleScheduledHTTP_RequestCancelDoesNotCancelTask(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + type ctxKey struct{} + type observation struct { + err error + value any + hasDeadline bool + deadline time.Time + } + observed := make(chan observation, 1) + + app := &Application{ + Purchase: &testutil.MockPurchaseManager{ + ProcessScheduledPurchasesFunc: func(ctx context.Context) (*purchase.ProcessResult, error) { + close(started) + <-release // the caller cancels the request context meanwhile + deadline, ok := ctx.Deadline() + observed <- observation{err: ctx.Err(), value: ctx.Value(ctxKey{}), hasDeadline: ok, deadline: deadline} + return &purchase.ProcessResult{Processed: 1, Executed: 1}, nil + }, + }, + } + + // Request-scoped values (request IDs, trace spans) must survive the detach. + reqCtx, cancelReq := context.WithCancel(context.WithValue(context.Background(), ctxKey{}, "request-scoped")) + defer cancelReq() + req := httptest.NewRequestWithContext(reqCtx, http.MethodPost, "/api/scheduled/process_scheduled_purchases", nil) + w := httptest.NewRecorder() + + done := make(chan struct{}) + go func() { + defer close(done) + app.handleScheduledHTTP(w, req) + }() + + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("scheduled task never started") + } + cancelReq() // scheduler disconnects mid-task + close(release) + + var got observation + select { + case got = <-observed: + case <-time.After(5 * time.Second): + t.Fatal("scheduled task never finished") + } + <-done + + testutil.AssertNoError(t, got.err) + testutil.AssertEqual(t, any("request-scoped"), got.value) + testutil.AssertTrue(t, got.hasDeadline, "detached task context must still carry a deadline") + remaining := time.Until(got.deadline) + testutil.AssertTrue(t, remaining > 0 && remaining <= scheduledTaskHTTPTimeout, + "task deadline must be bounded by scheduledTaskHTTPTimeout") + testutil.AssertEqual(t, http.StatusOK, w.Code) +} + +// TestScheduledHTTP_ResponseOutlivesServerWriteTimeout drives a real server +// built by CreateHTTPServer. Its 30s WriteTimeout is shorter than a sweep or +// exchange can take, so without a per-route write deadline a task that runs +// past it completes but the scheduler never receives the result (it sees a +// dropped connection and retries the same task). +func TestScheduledHTTP_ResponseOutlivesServerWriteTimeout(t *testing.T) { + app := &Application{ + Purchase: &testutil.MockPurchaseManager{ + ProcessScheduledPurchasesFunc: func(ctx context.Context) (*purchase.ProcessResult, error) { + time.Sleep(400 * time.Millisecond) // longer than the shortened WriteTimeout below + return &purchase.ProcessResult{Processed: 1}, nil + }, + }, + } + + srv := CreateHTTPServer(app, 0) + srv.WriteTimeout = 100 * time.Millisecond // stands in for the 30s production value + + ln, err := net.Listen("tcp", "127.0.0.1:0") + testutil.AssertNoError(t, err) + go func() { _ = srv.Serve(ln) }() + t.Cleanup(func() { _ = srv.Close() }) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + "http://"+ln.Addr().String()+"/api/scheduled/process_scheduled_purchases", nil) + testutil.AssertNoError(t, err) + + resp, err := http.DefaultClient.Do(req) + testutil.AssertNoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + testutil.AssertNoError(t, err) + + testutil.AssertEqual(t, http.StatusOK, resp.StatusCode) + testutil.AssertContains(t, string(body), `"status":"success"`) +} + +// TestStartHTTPServerCancelsScheduledTaskOnShutdown pins the shutdown path of +// the detached task: on SIGTERM the server drains, and a scheduled task still +// running when the drain window ends is canceled and given the rest of the +// grace period to unwind before StartHTTPServer returns (and main closes the +// database under it). +func TestStartHTTPServerCancelsScheduledTaskOnShutdown(t *testing.T) { + old := shutdownGracePeriod + shutdownGracePeriod = 600 * time.Millisecond + t.Cleanup(func() { shutdownGracePeriod = old }) + + started := make(chan struct{}) + unwound := make(chan error, 1) + app := &Application{ + Purchase: &testutil.MockPurchaseManager{ + ProcessScheduledPurchasesFunc: func(ctx context.Context) (*purchase.ProcessResult, error) { + close(started) + select { + case <-ctx.Done(): + unwound <- ctx.Err() + return nil, ctx.Err() + case <-time.After(10 * time.Second): + unwound <- nil + return &purchase.ProcessResult{}, nil + } + }, + }, + } + + l, err := net.Listen("tcp", "127.0.0.1:0") + testutil.AssertNoError(t, err) + port := l.Addr().(*net.TCPAddr).Port + testutil.AssertNoError(t, l.Close()) + + returned := make(chan error, 1) + go func() { returned <- StartHTTPServer(app, port) }() + + addr := fmt.Sprintf("127.0.0.1:%d", port) + deadline := time.Now().Add(5 * time.Second) + for { + conn, dialErr := net.DialTimeout("tcp", addr, 100*time.Millisecond) + if dialErr == nil { + _ = conn.Close() + break + } + if time.Now().After(deadline) { + t.Fatalf("server never started listening on %s: %v", addr, dialErr) + } + time.Sleep(10 * time.Millisecond) + } + + go func() { + resp, postErr := http.Post("http://"+addr+"/api/scheduled/process_scheduled_purchases", "application/json", nil) + if postErr == nil { + _ = resp.Body.Close() + } + }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("scheduled task never started") + } + sendSIGTERM(t) + + select { + case err := <-returned: + testutil.AssertNoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("StartHTTPServer did not return after the shutdown grace period") + } + // The task must have observed cancellation before StartHTTPServer returned. + select { + case err := <-unwound: + testutil.AssertTrue(t, err != nil, "scheduled task ran to completion instead of being canceled at shutdown") + default: + t.Fatal("StartHTTPServer returned while the detached scheduled task was still running") + } +} + +// TestScheduledTaskTimeoutFitsCloudRunRequestTimeout keeps the detached task +// bound inside the Cloud Run request_timeout configured in Terraform. Cloud Run +// runs with cpu_idle = true (module default cpu_throttling), so once the +// platform considers the request finished the instance's CPU is throttled and +// it may be scaled in; a bound beyond request_timeout would only promise time +// the platform does not give. +func TestScheduledTaskTimeoutFitsCloudRunRequestTimeout(t *testing.T) { + root := filepath.Join("..", "..", "terraform", "environments", "gcp") + files, err := filepath.Glob(filepath.Join(root, "*.tfvars*")) + testutil.AssertNoError(t, err) + files = append(files, filepath.Join(root, "variables.tf")) + + tfvar := regexp.MustCompile(`(?m)^\s*cloud_run_request_timeout\s*=\s*(\d+)`) + tfdefault := regexp.MustCompile(`(?s)variable "cloud_run_request_timeout" \{.*?default\s*=\s*(\d+)`) + checked := 0 + for _, f := range files { + data, readErr := os.ReadFile(f) // #nosec G304 -- fixed repo-relative paths + testutil.AssertNoError(t, readErr) + matches := tfvar.FindAllSubmatch(data, -1) + matches = append(matches, tfdefault.FindAllSubmatch(data, -1)...) + for _, m := range matches { + secs, convErr := strconv.Atoi(string(m[1])) + testutil.AssertNoError(t, convErr) + limit := time.Duration(secs) * time.Second + testutil.AssertTrue(t, scheduledTaskHTTPTimeout <= limit, + fmt.Sprintf("%s: scheduledTaskHTTPTimeout %v exceeds cloud_run_request_timeout %v", f, scheduledTaskHTTPTimeout, limit)) + checked++ + } + } + testutil.AssertTrue(t, checked >= 2, "expected to find cloud_run_request_timeout settings in terraform/environments/gcp") +} diff --git a/internal/server/scheduled_tasks.go b/internal/server/scheduled_tasks.go new file mode 100644 index 000000000..394b13209 --- /dev/null +++ b/internal/server/scheduled_tasks.go @@ -0,0 +1,108 @@ +package server + +import ( + "context" + "errors" + "log" + "net/http" + "sync" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/ctxutil" +) + +// scheduledTaskHTTPTimeout bounds a scheduled task triggered over HTTP. The +// task runs detached from the request context (#784), so this is what stops +// it. It matches the Cloud Run request_timeout configured in Terraform (300s +// in terraform/environments/gcp/*.tfvars and the variable default): Cloud Run +// runs with cpu_idle = true, so past that point the platform treats the +// request as finished, throttles the instance's CPU and may scale it in, and +// a longer bound would promise time the platform does not give. It also +// covers the purchase manager's own detached budget (4m execution + 30s +// terminal save). Azure Container Apps keeps CPU allocated after its ingress +// gives up (240s), so the same bound applies there. Pinned against the +// Terraform values by TestScheduledTaskTimeoutFitsCloudRunRequestTimeout. +const scheduledTaskHTTPTimeout = 300 * time.Second + +// scheduledResponseGrace is the extra time past scheduledTaskHTTPTimeout +// allowed for writing the response. +const scheduledResponseGrace = 30 * time.Second + +// extendScheduledWriteDeadline lifts the server-wide WriteTimeout (30s) for +// this response, so a task that runs longer can still report its result to a +// scheduler that is waiting for it instead of the connection being dropped. +func extendScheduledWriteDeadline(w http.ResponseWriter) { + deadline := time.Now().Add(scheduledTaskHTTPTimeout + scheduledResponseGrace) + if err := http.NewResponseController(w).SetWriteDeadline(deadline); err != nil && !errors.Is(err, http.ErrNotSupported) { + log.Printf("Failed to extend scheduled task write deadline: %v", err) + } +} + +// scheduledTaskTracker gives detached HTTP-triggered scheduled tasks a +// shutdown path. Detaching drops the request's cancellation, so without this a +// task would keep running after server shutdown and main would close the +// database under it. The zero value is ready to use. +type scheduledTaskTracker struct { + mu sync.Mutex + wg sync.WaitGroup + stopped bool + stopCh chan struct{} +} + +// start returns the context a scheduled task runs on: parent's values, no +// parent cancellation, bounded by d, and canceled by stop. Call done when the +// task returns. A task started after stop gets an already-canceled context. +func (t *scheduledTaskTracker) start(parent context.Context, d time.Duration) (ctx context.Context, done func()) { + ctx, cancel := ctxutil.DetachedWithTimeout(parent, d) + + t.mu.Lock() + if t.stopCh == nil { + t.stopCh = make(chan struct{}) + } + if t.stopped { + t.mu.Unlock() + cancel() + return ctx, func() {} + } + t.wg.Add(1) // under mu, so it cannot race with stop's Wait + stopCh := t.stopCh + t.mu.Unlock() + + go func() { + select { + case <-stopCh: + cancel() + case <-ctx.Done(): + } + }() + return ctx, func() { + cancel() + t.wg.Done() + } +} + +// stop cancels every in-flight task and waits for them to return until ctx is +// done. It reports whether all of them returned in time. +func (t *scheduledTaskTracker) stop(ctx context.Context) bool { + t.mu.Lock() + if t.stopCh == nil { + t.stopCh = make(chan struct{}) + } + if !t.stopped { + t.stopped = true + close(t.stopCh) + } + t.mu.Unlock() + + finished := make(chan struct{}) + go func() { + t.wg.Wait() + close(finished) + }() + select { + case <-finished: + return true + case <-ctx.Done(): + return false + } +}