Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 13 additions & 6 deletions internal/api/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ import (
"github.com/LeanerCloud/cloud-commitments-platform/internal/credentials"
"github.com/LeanerCloud/cloud-commitments-platform/internal/email"
"github.com/LeanerCloud/cloud-commitments-platform/internal/oidc"
"github.com/LeanerCloud/cloud-commitments-platform/internal/runtime"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-sdk-go-v2/aws"
awsconfig "github.com/aws/aws-sdk-go-v2/config"
Expand All @@ -44,6 +43,7 @@ type Handler struct {
signer oidc.Signer // Optional: OIDC issuer signer (backed by cloud KMS)
insurance InsuranceProvider // Optional: Archera comparison (nil = off)
issuerURL string // Canonical OIDC issuer URL (falls back to dashboardURL / request domain)
isLambda bool // Lambda mode from HandlerConfig.IsLambda; gates background goroutines

awsCfgOnce sync.Once // guards one-time loading of the base AWS config
awsCfg aws.Config // cached base AWS config (no region override)
Expand Down Expand Up @@ -148,17 +148,23 @@ type Handler struct {

// getRIUtilizationCache returns the Postgres-backed TTL cache for Cost
// Explorer results, lazy-initialized on first call so tests that never
// exercise the RI Exchange paths don't need to wire it up. Lambda
// detection happens here (once) via runtime.IsLambda so SWR is gated
// off on Lambda where background goroutines freeze between
// invocations.
// exercise the RI Exchange paths don't need to wire it up. The Lambda
// mode comes from h.isLambda (HandlerConfig.IsLambda, i.e. the
// Application's captured value, issue #110) so SWR is gated off on
// Lambda where background goroutines freeze between invocations.
func (h *Handler) getRIUtilizationCache() *riUtilizationCache {
h.riUtilizationCacheOnce.Do(func() {
h.riUtilizationCache = newRIUtilizationCache(h.config, runtime.IsLambda())
h.riUtilizationCache = newRIUtilizationCache(h.config, h.isLambda)
})
return h.riUtilizationCache
}

// IsLambda reports the Lambda mode this handler was constructed with
// (HandlerConfig.IsLambda). It never re-reads the environment.
func (h *Handler) IsLambda() bool {
return h.isLambda
}

// NewHandler creates a new API handler.
func NewHandler(cfg HandlerConfig) *Handler {
corsOrigin := cfg.CORSAllowedOrigin
Expand Down Expand Up @@ -190,6 +196,7 @@ func NewHandler(cfg HandlerConfig) *Handler {
insurance: cfg.Insurance,
commitmentOpts: cfg.CommitmentOpts,
encryptionKeySource: cfg.EncryptionKeySource,
isLambda: cfg.IsLambda,
}

// Pre-load API key (with a 5s timeout to avoid stalling cold-start indefinitely)
Expand Down
14 changes: 7 additions & 7 deletions internal/api/handler_history.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@ import (
"github.com/LeanerCloud/cloud-commitments-go/pkg/logging"
"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
"github.com/LeanerCloud/cloud-commitments-platform/internal/purchase"
"github.com/LeanerCloud/cloud-commitments-platform/internal/runtime"
"github.com/aws/aws-lambda-go/events"
"github.com/jackc/pgx/v5"
)
Expand Down Expand Up @@ -247,11 +246,12 @@ func isStaleExecution(exec config.PurchaseExecution) bool {
// 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 SWR cache's isLambda gate (ri_utilization_cache.go) and reading the same
// h.isLambda value (HandlerConfig.IsLambda, the Application's single captured
// answer, issue #110) rather than the live environment. 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 sweep is idempotent per execution ID: TransitionExecutionStatus is
// guarded by the FROM-status list ("pending","notified"), so a concurrent
Expand All @@ -267,7 +267,7 @@ func (h *Handler) expireStaleExecutions(staleExecs []config.PurchaseExecution) {
logging.Warnf("history: sweeping %d of %d stale executions", config.DefaultListLimit, len(staleExecs))
staleExecs = staleExecs[:config.DefaultListLimit]
}
if runtime.IsLambda() {
if h.isLambda {
h.expireStaleExecutionsSweep(staleExecs)
return
}
Expand Down
7 changes: 5 additions & 2 deletions internal/api/handler_history_actionable_starvation_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -158,14 +158,17 @@ func TestHandler_getHistory_TruncatedPerClass(t *testing.T) {
}

func TestHandler_getHistory_StaleSweepBoundedToDefaultListLimit(t *testing.T) {
t.Setenv("AWS_LAMBDA_RUNTIME_API", "127.0.0.1:9001")
ctx := context.Background()
stale := executionsWithStatus(150, "pending", time.Now().Add(-8*24*time.Hour), "stale")
store := newStarvationStore(ctx)
store.On("GetExecutionsByStatuses", ctx, mock.Anything, mock.Anything).Return(stale, nil)
store.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, mock.Anything).Return(false, nil)

runStarvationHistory(t, ctx, store)
// Lambda mode (synchronous sweep) so the call count is final on return.
mockAuth, req := adminHistoryReq(ctx)
handler := &Handler{auth: mockAuth, config: store, isLambda: true}
_, err := handler.getHistory(ctx, req, map[string]string{})
require.NoError(t, err)

store.AssertNumberOfCalls(t, "ExpireExecutionAtomic", config.DefaultListLimit)
}
Expand Down
24 changes: 13 additions & 11 deletions internal/api/handler_history_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1942,9 +1942,11 @@ func TestSummarizePurchaseHistory_CancelPendingDoesNotChangeKPIs(t *testing.T) {
// 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.
// the transitions. The mode comes from Handler.isLambda (HandlerConfig.IsLambda,
// the Application's single captured value), the same field the SWR cache
// goroutine gate reads. Each sub-test sets AWS_LAMBDA_RUNTIME_API to the
// opposite of the handler's mode to prove the live environment is ignored
// (issue #110).
func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
approverEmail := "ops@example.com"
staleID := "stale-exec-lambda-guard"
Expand All @@ -1962,9 +1964,9 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
}

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")
// The environment says "not Lambda"; the handler's configured mode
// must win (issue #110).
t.Setenv("AWS_LAMBDA_RUNTIME_API", "")

ctx := context.Background()
mockStore := new(MockConfigStore)
Expand All @@ -1979,7 +1981,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
mockStore.On("ExpireExecutionAtomic", mock.Anything, mock.Anything, staleID).Return(true, nil).Once()

mockAuth, req := adminHistoryReq(ctx)
handler := &Handler{auth: mockAuth, config: mockStore}
handler := &Handler{auth: mockAuth, config: mockStore, isLambda: true}

result, err := handler.getHistory(ctx, req, map[string]string{})
require.NoError(t, err)
Expand All @@ -1996,9 +1998,9 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
})

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", "")
// The environment says "Lambda"; the handler's configured mode must
// win (issue #110). t.Setenv restores the variable afterwards.
t.Setenv("AWS_LAMBDA_RUNTIME_API", "127.0.0.1:9001")

ctx := context.Background()
mockStore := new(MockConfigStore)
Expand All @@ -2021,7 +2023,7 @@ func TestHandler_getHistory_ExpireIfStale_LambdaGuard(t *testing.T) {
}).Return(true, nil).Once()

mockAuth, req := adminHistoryReq(ctx)
handler := &Handler{auth: mockAuth, config: mockStore}
handler := &Handler{auth: mockAuth, config: mockStore, isLambda: false}

done := make(chan struct{})
var result any
Expand Down
25 changes: 25 additions & 0 deletions internal/api/ri_utilization_cache_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"sync/atomic"
"testing"
Expand Down Expand Up @@ -330,3 +331,27 @@ func TestRIUtilizationCache_CorruptPayloadFallsThrough(t *testing.T) {
t.Fatalf("stored payload still corrupt after upsert: %v", err)
}
}

// TestHandler_getRIUtilizationCache_FollowsHandlerLambdaMode guards issue
// #110: the SWR gate must take the handler's configured Lambda mode
// (HandlerConfig.IsLambda), not a live read of AWS_LAMBDA_RUNTIME_API. Each
// case sets the environment to the opposite of the configured mode.
func TestHandler_getRIUtilizationCache_FollowsHandlerLambdaMode(t *testing.T) {
for _, isLambda := range []bool{true, false} {
t.Run(fmt.Sprintf("isLambda=%t", isLambda), func(t *testing.T) {
env := ""
if !isLambda {
env = "127.0.0.1:9001"
}
t.Setenv("AWS_LAMBDA_RUNTIME_API", env)

h := NewHandler(HandlerConfig{CORSAllowedOrigin: "https://islambda.example.test", IsLambda: isLambda})
if got := h.IsLambda(); got != isLambda {
t.Fatalf("Handler.IsLambda() = %t, want %t", got, isLambda)
}
if got := h.getRIUtilizationCache().isLambda; got != isLambda {
t.Fatalf("RI utilization cache isLambda = %t, want %t (must follow HandlerConfig.IsLambda)", got, isLambda)
}
})
}
}
5 changes: 5 additions & 0 deletions internal/api/types.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,11 @@ type HandlerConfig struct {
APIKeySecretARN string
EncryptionKeySource string
EnableDashboard bool
// IsLambda is the caller's (Application's) single answer to "am I on
// Lambda". It gates the background goroutines (RI utilization SWR
// refresh, stale-execution sweep) that would freeze mid-flight on Lambda.
// The handler never re-reads the environment itself (issue #110).
IsLambda bool
}

// CommitmentOptsInterface lets us swap the real *commitmentopts.Service for
Expand Down
8 changes: 8 additions & 0 deletions internal/scheduler/scheduler.go
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,14 @@ func NewScheduler(cfg SchedulerConfig) *Scheduler {
}
}

// IsLambda reports the Lambda mode this scheduler was constructed with
// (SchedulerConfig.IsLambda). It never re-reads the environment: the value is
// owned by whoever built the scheduler, so callers can assert that the mode
// they threaded in is the one that gates background goroutines (issue #110).
func (s *Scheduler) IsLambda() bool {
return s.isLambda
}

// cacheTTLFromEnv parses CUDLY_RECOMMENDATION_CACHE_TTL using
// time.ParseDuration. Falls back to defaultCacheTTL on parse error (with
// a warning) so a mis-set env doesn't take the feature offline.
Expand Down
7 changes: 6 additions & 1 deletion internal/server/app.go
Original file line number Diff line number Diff line change
Expand Up @@ -491,7 +491,10 @@ func NewApplicationFromDeps(ctx context.Context, cfg ApplicationConfig, deps Ext
PurchaseManager: purchaseManager,
EmailSender: deps.EmailSender,
STSClient: deps.STSClient,
IsLambda: runtime.IsLambda(),
// cfg.IsLambda, never a live runtime.IsLambda(): the Application has
// exactly one answer to "am I on Lambda", shared with the post-DB
// scheduler in ensureDB and the API handler (issue #110).
IsLambda: cfg.IsLambda,
})
purchaseManager.SetRecommendationSource(sched)

Expand Down Expand Up @@ -531,6 +534,7 @@ func NewApplicationFromDeps(ctx context.Context, cfg ApplicationConfig, deps Ext
OIDCSigner: signer,
OIDCIssuerURL: resolveOIDCIssuerURL(cfg),
Insurance: archeraProvider,
IsLambda: cfg.IsLambda,
})

log.Printf("CUDly Server initialization complete")
Expand Down Expand Up @@ -899,6 +903,7 @@ func (app *Application) reinitializeAfterConnect(ctx context.Context, dbConn *da
CommitmentOpts: commitmentOpts,
EncryptionKeySource: app.encKeySource,
Insurance: app.archera,
IsLambda: app.appConfig.IsLambda,
})
if app.API == nil {
return fmt.Errorf("failed to create API handler")
Expand Down
67 changes: 67 additions & 0 deletions internal/server/app_islambda_integration_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
//go:build integration

package server

import (
"context"
"strings"
"testing"
"time"

"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
"github.com/stretchr/testify/require"
)

// TestEnsureDB_IsLambdaSingleSourceOfTruth extends the issue #110 regression
// to the post-DB wiring: after ensureDB rebuilds the scheduler and API handler,
// both must still follow ApplicationConfig.IsLambda while the environment says
// the opposite.
func TestEnsureDB_IsLambdaSingleSourceOfTruth(t *testing.T) {
for _, key := range []string{
"CREDENTIAL_ENCRYPTION_KEY_SECRET_ARN", "CREDENTIAL_ENCRYPTION_KEY_SECRET_NAME",
"CREDENTIAL_ENCRYPTION_KEY_SECRET_ID", "CREDENTIAL_ENCRYPTION_ALLOW_DEV_KEY",
"CUDLY_SIGNING_KEY_ID", "CUDLY_SIGNING_KEY_VAULT_URL", "CUDLY_SIGNING_KEY_NAME",
"CUDLY_SIGNING_KEY_RESOURCE", "AWS_PROFILE",
} {
t.Setenv(key, "")
}
t.Setenv("CREDENTIAL_ENCRYPTION_KEY", strings.Repeat("12", 32))
t.Setenv("CUDLY_ISSUER_URL", "https://islambda.example.test")
t.Setenv("CUDLY_SOURCE_CLOUD", "aws")
t.Setenv("SCHEDULED_TASK_AUTH_MODE", "disabled")
t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
t.Setenv("AWS_ACCESS_KEY_ID", "local-islambda-test")
t.Setenv("AWS_SECRET_ACCESS_KEY", "local-islambda-test")
t.Setenv("AWS_REGION", "us-east-1")

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
pg, err := testhelpers.SetupPostgresContainer(ctx, t)
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, pg.Cleanup(context.Background())) })
require.NoError(t, migrations.RunMigrations(ctx, pg.DB.Pool(), "../database/postgres/migrations", "", ""))

for _, cfgIsLambda := range []bool{true, false} {
name := "config not Lambda, environment Lambda"
if cfgIsLambda {
name = "config Lambda, environment not Lambda"
}
t.Run(name, func(t *testing.T) {
setLambdaEnv(t, !cfgIsLambda)
cfg := ApplicationConfig{
IsLambda: cfgIsLambda, DashboardURL: "https://islambda.example.test",
DefaultTerm: 3, DefaultCoverage: 80,
}
app, appErr := NewApplicationFromDeps(ctx, cfg, ExternalDeps{
DBConfig: pg.Config, EmailSender: &noopEmailSender{},
})
require.NoError(t, appErr)
t.Cleanup(func() { require.NoError(t, app.Close()) })

require.NoError(t, app.ensureDB(ctx))
require.NotNil(t, app.DB, "ensureDB must have rebuilt the post-DB wiring")
assertLambdaModeWiring(t, app, cfgIsLambda)
})
}
}
Loading
Loading