fix(server): detach HTTP-triggered scheduled tasks from the request context

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)
+}

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")
+	}
+}

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.

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)

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")
+}