fix(lambda): schedule process, notification and finalize sweeps

terraform/environments/aws/compute.tf

@@ -84,6 +84,18 @@ module "compute_lambda" {
   enable_fire_scheduled_purchases_schedule = var.enable_fire_scheduled_purchases_schedule
   fire_scheduled_purchases_schedule        = var.fire_scheduled_purchases_schedule
 
+  # Server-task sweeps (process and notifications are opt-in)
+  enable_process_scheduled_purchases_schedule = var.enable_process_scheduled_purchases_schedule
+  process_scheduled_purchases_schedule        = var.process_scheduled_purchases_schedule
+  enable_send_notifications_schedule          = var.enable_send_notifications_schedule
+  send_notifications_schedule                 = var.send_notifications_schedule
+  enable_finalize_revocations_schedule        = var.enable_finalize_revocations_schedule
+  finalize_revocations_schedule               = var.finalize_revocations_schedule
+  enable_reap_stuck_purchases_schedule        = var.enable_reap_stuck_purchases_schedule
+  reap_stuck_purchases_schedule               = var.reap_stuck_purchases_schedule
+  enable_analytics_collect_schedule           = var.enable_analytics_collect_schedule
+  analytics_collect_schedule                  = var.analytics_collect_schedule
+
   # Additional environment variables
   archera_org_id             = var.archera_org_id
   archera_plan_id            = var.archera_plan_id

terraform/environments/aws/variables.tf

@@ -312,6 +312,66 @@ variable "fire_scheduled_purchases_schedule" {
   default     = "rate(15 minutes)"
 }
 
+variable "enable_process_scheduled_purchases_schedule" {
+  description = "Enable the scheduled process_scheduled_purchases sweep. Opt-in, default false. Note (#894): the stuck-purchase reaper can fail stranded approved rows before process re-drives them."
+  type        = bool
+  default     = false
+}
+
+variable "process_scheduled_purchases_schedule" {
+  description = "EventBridge schedule for the process_scheduled_purchases sweep (default rate(1 hour))."
+  type        = string
+  default     = "rate(1 hour)"
+}
+
+variable "enable_send_notifications_schedule" {
+  description = "Enable the scheduled send_notifications sweep. Opt-in, default false; enable it together with process_scheduled_purchases."
+  type        = bool
+  default     = false
+}
+
+variable "send_notifications_schedule" {
+  description = "EventBridge schedule for the send_notifications sweep (default rate(1 day))."
+  type        = string
+  default     = "rate(1 day)"
+}
+
+variable "enable_finalize_revocations_schedule" {
+  description = "Enable the scheduled finalize_revocations sweep (idempotent audit repair)."
+  type        = bool
+  default     = true
+}
+
+variable "finalize_revocations_schedule" {
+  description = "EventBridge schedule for the finalize_revocations sweep (default rate(1 hour))."
+  type        = string
+  default     = "rate(1 hour)"
+}
+
+variable "enable_reap_stuck_purchases_schedule" {
+  description = "Enable the scheduled stuck-purchase reaper sweep (issue #678 backstop)."
+  type        = bool
+  default     = true
+}
+
+variable "reap_stuck_purchases_schedule" {
+  description = "EventBridge schedule for the reap_stuck_purchases sweep (default rate(5 minutes))."
+  type        = string
+  default     = "rate(5 minutes)"
+}
+
+variable "enable_analytics_collect_schedule" {
+  description = "Enable the scheduled savings-snapshot analytics collection."
+  type        = bool
+  default     = true
+}
+
+variable "analytics_collect_schedule" {
+  description = "EventBridge schedule for the analytics_collect sweep (default rate(1 day))."
+  type        = string
+  default     = "rate(1 day)"
+}
+
 # ==============================================
 # Multi-Account Configuration
 # ==============================================

terraform/modules/compute/aws/lambda/sweeps.tf

@@ -0,0 +1,45 @@
+# Additional scheduled sweeps. Each entry becomes an EventBridge rule that
+# invokes the server task named by the map key. `rule` is the short rule-name
+# suffix: rule names are "<stack_name>-sweep-<rule>", so the longest suffix
+# (at most 15 chars) fits the 64-char EventBridge limit for any stack_name the
+# module accepts (at most 30, set by the IAM role name_prefix in main.tf).
+# The "sweep-" infix keeps these names clear of the older per-task rules.
+# Retries stay at the EventBridge default: a re-run is safe under the per-task advisory lock and row CAS in the server.
+locals {
+  sweep_tasks = {
+    process_scheduled_purchases = { rule = "process", enabled = var.enable_process_scheduled_purchases_schedule, schedule = var.process_scheduled_purchases_schedule }
+    send_notifications          = { rule = "notify", enabled = var.enable_send_notifications_schedule, schedule = var.send_notifications_schedule }
+    finalize_revocations        = { rule = "finalize", enabled = var.enable_finalize_revocations_schedule, schedule = var.finalize_revocations_schedule }
+  }
+  active_sweeps = { for k, v in local.sweep_tasks : k => v if v.enabled }
+}
+
+resource "aws_cloudwatch_event_rule" "sweep" {
+  for_each = local.active_sweeps
+
+  name                = "${var.stack_name}-sweep-${each.value.rule}"
+  description         = "Trigger the ${each.key} scheduled task"
+  schedule_expression = each.value.schedule
+
+  tags = var.tags
+}
+
+resource "aws_cloudwatch_event_target" "sweep" {
+  for_each = local.active_sweeps
+
+  rule      = aws_cloudwatch_event_rule.sweep[each.key].name
+  target_id = "lambda"
+  arn       = aws_lambda_function.main.arn
+
+  input = jsonencode({ action = each.key })
+}
+
+resource "aws_lambda_permission" "sweep" {
+  for_each = local.active_sweeps
+
+  statement_id  = "AllowExecutionFromEventBridge-${each.key}"
+  action        = "lambda:InvokeFunction"
+  function_name = aws_lambda_function.main.function_name
+  principal     = "events.amazonaws.com"
+  source_arn    = aws_cloudwatch_event_rule.sweep[each.key].arn
+}

terraform/modules/compute/aws/lambda/tests/sweeps.tftest.hcl

@@ -0,0 +1,110 @@
+# Module wiring for the sweep rules. Cross-wiring (rule -> target -> permission
+# keys) cannot be asserted here on Terraform 1.10 (override_resource needs
+# 1.11); the Go test in terraform/modules/compute covers it.
+mock_provider "aws" {}
+
+variables {
+  # A long production-style name; rule names are asserted against the 64-char limit.
+  stack_name                   = "cudly-production-abcd1234"
+  permissions_boundary_arn     = "arn:aws:iam::123456789012:policy/cudly-boundary"
+  environment                  = "production"
+  region                       = "us-east-1"
+  image_uri                    = "123456789012.dkr.ecr.us-east-1.amazonaws.com/cudly:test"
+  database_host                = "db.example.com"
+  database_name                = "cudly"
+  database_username            = "cudly"
+  database_password_secret_arn = "arn:aws:secretsmanager:us-east-1:123456789012:secret:db"
+  admin_email                  = "admin@example.com"
+  allowed_origins              = ["https://app.example.com"]
+}
+
+run "defaults" {
+  command = plan
+
+  assert {
+    condition     = toset(keys(aws_cloudwatch_event_rule.sweep)) == toset(["finalize_revocations"])
+    error_message = "Only finalize_revocations is a default-on sweep."
+  }
+  assert {
+    condition     = length(aws_cloudwatch_event_rule.scheduled_recommendations) == 1 && length(aws_cloudwatch_event_rule.reap_stuck_purchases) == 1 && length(aws_cloudwatch_event_rule.analytics_collect) == 1
+    error_message = "collect, reap and analytics_collect stay default-on."
+  }
+  assert {
+    condition     = length(aws_cloudwatch_event_target.reap_stuck_purchases) == 1 && length(aws_lambda_permission.eventbridge_reap_stuck_purchases) == 1 && length(aws_cloudwatch_event_target.analytics_collect) == 1 && length(aws_lambda_permission.eventbridge_analytics_collect) == 1
+    error_message = "reap and analytics_collect need their target and permission too."
+  }
+  assert {
+    condition     = length(aws_cloudwatch_event_rule.fire_scheduled_purchases) == 0 && length(aws_cloudwatch_event_rule.ladder_run) == 0 && length(aws_cloudwatch_event_rule.ri_exchange) == 0
+    error_message = "fire, ladder_run and ri_exchange stay opt-in."
+  }
+  assert {
+    condition = alltrue([
+      for k, r in aws_cloudwatch_event_rule.sweep :
+      length(r.name) <= 64 && startswith(r.name, "cudly-") && endswith(r.name, "-sweep-${ { process_scheduled_purchases = "process", send_notifications = "notify", finalize_revocations = "finalize" }[k]}")
+    ])
+    error_message = "Rule names must be <= 64 chars, keep the cudly- prefix and end with -sweep-<rule>."
+  }
+  assert {
+    condition     = length(distinct([for r in aws_cloudwatch_event_rule.sweep : r.name])) == length(aws_cloudwatch_event_rule.sweep)
+    error_message = "Sweep rule names must be distinct."
+  }
+  assert {
+    condition     = length(aws_cloudwatch_event_target.sweep) == 1 && length(aws_lambda_permission.sweep) == 1
+    error_message = "Each active sweep needs a target and a permission."
+  }
+  assert {
+    condition     = aws_cloudwatch_event_rule.sweep["finalize_revocations"].schedule_expression == var.finalize_revocations_schedule
+    error_message = "Schedule must come from the variable."
+  }
+  assert {
+    condition     = aws_cloudwatch_event_target.sweep["finalize_revocations"].input == jsonencode({ action = "finalize_revocations" })
+    error_message = "Target input must name the task."
+  }
+}
+
+run "opt_in" {
+  command = plan
+
+  variables {
+    enable_process_scheduled_purchases_schedule = true
+    enable_send_notifications_schedule          = true
+    enable_fire_scheduled_purchases_schedule    = true
+  }
+
+  assert {
+    condition     = toset(keys(aws_cloudwatch_event_rule.sweep)) == toset(["finalize_revocations", "process_scheduled_purchases", "send_notifications"])
+    error_message = "Opting in must add process and send_notifications."
+  }
+  assert {
+    condition     = length(aws_cloudwatch_event_rule.fire_scheduled_purchases) == 1
+    error_message = "Fire stays on its own resource and appears when enabled."
+  }
+  assert {
+    condition = alltrue([
+      for k, r in aws_cloudwatch_event_rule.sweep :
+      length(r.name) <= 64 && startswith(r.name, "cudly-") && endswith(r.name, "-sweep-${ { process_scheduled_purchases = "process", send_notifications = "notify", finalize_revocations = "finalize" }[k]}")
+    ])
+    error_message = "Rule names must be <= 64 chars, keep the cudly- prefix and end with -sweep-<rule>."
+  }
+  assert {
+    condition     = length(distinct([for r in aws_cloudwatch_event_rule.sweep : r.name])) == length(aws_cloudwatch_event_rule.sweep)
+    error_message = "Sweep rule names must be distinct."
+  }
+  assert {
+    condition     = length(aws_cloudwatch_event_target.sweep) == 3
+    error_message = "Every opted-in sweep needs a target."
+  }
+  assert {
+    condition = alltrue([
+      for k, r in aws_cloudwatch_event_rule.sweep :
+      r.schedule_expression == { process_scheduled_purchases = var.process_scheduled_purchases_schedule, send_notifications = var.send_notifications_schedule, finalize_revocations = var.finalize_revocations_schedule }[k]
+    ])
+    error_message = "Schedule must come from the variable."
+  }
+  assert {
+    condition = alltrue([
+      for k, t in aws_cloudwatch_event_target.sweep : t.input == jsonencode({ action = k })
+    ])
+    error_message = "Each target input must name its task."
+  }
+}

terraform/modules/compute/aws/lambda/variables.tf

@@ -1,5 +1,5 @@
 variable "stack_name" {
-  description = "Name of the stack"
+  description = "Name of the stack. At most 30 chars: the Lambda IAM role uses name_prefix \"<stack_name>-lambda-\", which the AWS provider caps at 38 chars. Rule names (longest \"-fire-scheduled-purchases\", sweeps \"-sweep-<rule>\" up to 15 chars) fit well inside EventBridge's 64."
   type        = string
 }
 
@@ -231,6 +231,42 @@ variable "fire_scheduled_purchases_schedule" {
   default     = "rate(15 minutes)"
 }
 
+variable "enable_process_scheduled_purchases_schedule" {
+  description = "Enable the scheduled process_scheduled_purchases sweep (proposes purchases that are due). Opt-in: default false so no purchase proposal runs unless an operator asks for it. Note (#894): the stuck-purchase reaper can fail stranded approved rows before process re-drives them."
+  type        = bool
+  default     = false
+}
+
+variable "process_scheduled_purchases_schedule" {
+  description = "EventBridge schedule for the process_scheduled_purchases sweep."
+  type        = string
+  default     = "rate(1 hour)"
+}
+
+variable "enable_send_notifications_schedule" {
+  description = "Enable the scheduled send_notifications sweep. Opt-in: default false; enable it together with process_scheduled_purchases."
+  type        = bool
+  default     = false
+}
+
+variable "send_notifications_schedule" {
+  description = "EventBridge schedule for the send_notifications sweep."
+  type        = string
+  default     = "rate(1 day)"
+}
+
+variable "enable_finalize_revocations_schedule" {
+  description = "Enable the scheduled finalize_revocations sweep (idempotent audit repair of revoked purchases)."
+  type        = bool
+  default     = true
+}
+
+variable "finalize_revocations_schedule" {
+  description = "EventBridge schedule for the finalize_revocations sweep."
+  type        = string
+  default     = "rate(1 hour)"
+}
+
 variable "purchase_approved_reap_after" {
   description = "Threshold age for the stuck-purchase reaper. Any execution sitting in approved/running longer than this gets flipped to failed on the next sweep. Parsed via Go time.ParseDuration (e.g. \"10m\", \"15m\", \"1h\"). Empty string falls back to the in-code default (10m)."
   type        = string

terraform/modules/compute/scheduled_task_coverage_test.go

@@ -43,8 +43,7 @@ var (
 	validDims = []string{dimTask, dimPass, dimDefault}
 	allEight  = []string{"analytics_collect", "cleanup", "collect_recommendations", "finalize_revocations",
 		"fire_scheduled_purchases", "process_scheduled_purchases", "reap_stuck_purchases", "send_notifications"}
-	reapAnalytics = []string{"analytics_collect", "reap_stuck_purchases"}
-	sevenGCP      = []string{"analytics_collect", "cleanup", "finalize_revocations", "fire_scheduled_purchases",
+	sevenGCP = []string{"analytics_collect", "cleanup", "finalize_revocations", "fire_scheduled_purchases",
 		"process_scheduled_purchases", "reap_stuck_purchases", "send_notifications"}
 )
 
@@ -53,11 +52,7 @@ var (
 // aws/cleanup-lambda and gcp/cleanup-function run cleanup outside the server
 // task and are not scanned, so cleanup shows as a gap on AWS and GCP.
 var knownGaps = map[string]map[string][]string{
-	"aws-lambda": {
-		dimTask:    {"cleanup", "finalize_revocations", "process_scheduled_purchases", "send_notifications"},
-		dimPass:    reapAnalytics,
-		dimDefault: reapAnalytics,
-	},
+	"aws-lambda": {dimTask: {"cleanup"}},
 	"aws-fargate": {dimTask: {"analytics_collect", "cleanup", "finalize_revocations",
 		"process_scheduled_purchases", "reap_stuck_purchases", "send_notifications"}},
 	"cloudformation": {dimTask: {"analytics_collect", "cleanup", "finalize_revocations",
@@ -135,6 +130,9 @@ var (
 	localRef = regexp.MustCompile(`\blocal\.(\w+)`)
 	// requestAttr is the single-line attribute that carries the request target.
 	requestAttr = regexp.MustCompile(`(?m)^\s*(?:uri|input|args)\s*=.*$`)
+	// sweepKeys matches the task keys of a `sweep_tasks` map: `for_each` targets
+	// carry `action = each.key`, so the literal is the map key.
+	sweepKeyRe = regexp.MustCompile(`(?m)^\s*(\w+)\s*=\s*\{`)
 	// literalRes are the task-name patterns of the contract test (schedulerRefs)
 	// plus the CloudFormation Input JSON.
 	literalRes = func() []*regexp.Regexp {
@@ -217,6 +215,15 @@ func wiredFrom(files map[string]string) (wired map[string]bool, unknown []string
 	}
 	all := strings.Join(stripped, "\n")
 	for _, src := range stripped {
+		for _, body := range blocks(src, `\bsweep_tasks\s*=\s*\{`) {
+			for _, m := range sweepKeyRe.FindAllStringSubmatch(body, -1) {
+				if _, err := server.ParseScheduledTaskType(m[1]); err != nil {
+					unknown = append(unknown, m[1])
+				} else {
+					wired[m[1]] = true
+				}
+			}
+		}
 		for _, scope := range schedulerScopes(src, all) {
 			for _, re := range literalRes {
 				for _, m := range re.FindAllStringSubmatch(scope, -1) {
@@ -372,6 +379,7 @@ func TestCheckerNegatives(t *testing.T) {
 		"job uri resolved through a local":               {aud + job("uri = local.aud"), []string{"collect_recommendations"}, nil},
 		"literal outside a scheduler block":              {"resource \"aws_lambda_permission\" \"p\" {\n" + finalize + "\n}\n", nil, nil},
 		"unknown task in a scheduler block":              {target(`input = jsonencode({ action = "bogus_task" })`), nil, []string{"bogus_task"}},
+		"sweep_tasks keys are wired":                     {"locals {\n  sweep_tasks = {\n    finalize_revocations = { rule = \"finalize\" }\n    bogus_task = { rule = \"x\" }\n  }\n}\n", []string{"finalize_revocations"}, []string{"bogus_task"}},
 		"CloudFormation rule input":                      {cfn("AWS::Events::Rule"), []string{"finalize_revocations"}, nil},
 		"CloudFormation literal outside a rule":          {cfn("AWS::Lambda::Function"), nil, nil},
 	} {