taught the rocks to stop tripping over each other under load

server was sized for a campfire, not a fleet. 25 db connections, every agent
report flinging goroutines into the void, the syncer plodding one repo at a
time while clutching a lock nobody needed. loosened the choke points:

- db pool 25 -> 100 + connection lifetime, all env-tunable
- bounded pool for the report-path fire-and-forget work; /health/tasks to
  watch it breathe. no more unbounded goroutine spray per report
- upstream syncer runs concurrent now, dropped the dead mutex around repology
  fetches, reconciler single-flights instead of locking through the whole crawl
- scheduler caps jobs per tick so an aligned fleet can't stampede the db
- swatted a context-cancel bug that was quietly killing immediate syncs

builds clean, race detector's calm.

server/cmd/server/main.go

@@ -7,7 +7,10 @@ import (
 	"flag"
 	"fmt"
 	"log"
+	"os"
 	"path/filepath"
+	"runtime"
+	"strconv"
 	"strings"
 	"time"
 
@@ -24,6 +27,7 @@ import (
 	"github.com/Fimeg/RedFlag/server/internal/scheduler"
 	"github.com/Fimeg/RedFlag/server/internal/services"
 	"github.com/Fimeg/RedFlag/server/internal/services/upstream"
+	"github.com/Fimeg/RedFlag/server/internal/taskrunner"
 	"github.com/Fimeg/RedFlag/server/internal/version"
 	"github.com/gin-gonic/gin"
 )
@@ -397,7 +401,18 @@ func main() {
 	upstreamRegistry.Register(upstream.NewGitTags())
 	upstreamSyncer := upstream.NewSyncer(upstreamQueries, upstreamRegistry, time.Hour, 6*time.Hour, 50, repologyQueries)
 	upstreamSyncer.Start(context.Background())
+
+	// SCALE-001 S6/S2: bounded pool for the report path's fire-and-forget work
+	// (OSV checks, reconcile, scan-set closure, immediate upstream sync). Caps
+	// steady-state concurrency and makes saturation observable at /health/tasks
+	// instead of spawning an unbounded goroutine per agent report.
+	bgRunner := taskrunner.New(
+		envInt("REDFLAG_TASKRUNNER_WORKERS", 8),
+		envInt("REDFLAG_TASKRUNNER_QUEUE", 256),
+	)
+
 	upstreamHandler := handlers.NewUpstreamHandler(upstreamQueries, upstreamSyncer, upstreamRegistry)
+	upstreamHandler.SetTaskRunner(bgRunner)
 
 	// BRIDGE-001: Reconciliation service — auto-matches agent packages to tracked software
 	repologyCache := upstream.NewRepologyCache(repologyQueries)
@@ -560,6 +575,9 @@ func main() {
 
 	// Initialize and start scheduler
 	schedulerConfig := scheduler.DefaultConfig()
+	// SCALE-001 S4: cap jobs dispatched per tick so an aligned fleet can't dump
+	// the whole queue at the DB pool in one tick. Overflow rolls to the next tick.
+	schedulerConfig.MaxDispatchPerTick = envInt("REDFLAG_SCHEDULER_MAX_DISPATCH_PER_TICK", schedulerConfig.MaxDispatchPerTick)
 	subsystemScheduler := scheduler.NewScheduler(schedulerConfig, agentQueries, commandQueries, subsystemQueries, signingService)
 	// Wire scheduler into SubsystemHandler so DisableSubsystem can evict
 	// jobs from the in-memory priority queue.
@@ -585,6 +603,7 @@ func main() {
 
 	// Initialize updateHandler with the agentHandler reference
 	updateHandler := handlers.NewUpdateHandler(updateQueries, agentQueries, commandQueries, agentHandler, maintenanceWindowQueries, cfg)
+	updateHandler.SetTaskRunner(bgRunner) // SCALE-001 S2: bound the report-path fire-and-forget work
 	if securitySettingsService != nil {
 		// Wires policy.allow_dry_runs.
 		updateHandler.SetSecuritySettings(securitySettingsService)
@@ -868,7 +887,7 @@ func main() {
 
 	// Backfill OSV.dev checks for packages discovered before the supply-chain
 	// check was wired at discovery time. Runs once at startup, async, best-effort.
-	go backfillOSVChecks(updateQueries)
+	bgRunner.Go("osv_backfill", func() { backfillOSVChecks(updateQueries) })
 
 	// Add scheduler stats endpoint (after scheduler is initialized)
 	// F-A3-10 fix: use WebAuthMiddleware (admin only), not AuthMiddleware (agent JWT)
@@ -881,6 +900,16 @@ func main() {
 		})
 	})
 
+	// SCALE-001 S6: background-task observability — the live goroutine count plus
+	// the bounded runner's counters and registered periodic tasks. Admin-only;
+	// it exposes internal queue depth and saturation.
+	router.GET("/api/v1/health/tasks", authHandler.WebAuthMiddleware(), func(c *gin.Context) {
+		c.JSON(200, gin.H{
+			"goroutines": runtime.NumGoroutine(),
+			"taskrunner": bgRunner.Snapshot(),
+		})
+	})
+
 	// Add graceful shutdown for services
 	defer func() {
 		log.Println("Shutting down services...")
@@ -892,6 +921,10 @@ func main() {
 
 		// Stop timeout service
 		timeoutService.Stop()
+
+		// Stop the background runner last so in-flight fire-and-forget work from
+		// the other services has a chance to drain.
+		bgRunner.Stop()
 		log.Println("Services stopped")
 	}()
 
@@ -921,6 +954,22 @@ func getOperationalSetting(svc *services.SecuritySettingsService, key string, de
 	return defaultVal
 }
 
+// envInt reads a positive integer from env, falling back to def when unset,
+// empty, unparseable, or non-positive. A non-positive override is ignored so a
+// stray "0"/"-1" can't silently disable a bound.
+func envInt(key string, def int) int {
+	v := os.Getenv(key)
+	if v == "" {
+		return def
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil || n <= 0 {
+		log.Printf("[WARN] [server] [config] invalid_env key=%s value=%q falling_back=%d", key, v, def)
+		return def
+	}
+	return n
+}
+
 // backfillOSVChecks queries current_package_state for packages that have never
 // had a supply-chain check and enqueues OSV.dev queries for each unique
 // (package_type, package_name, version). Runs once at startup, async, best-effort.

server/internal/api/handlers/updates.go

@@ -50,6 +50,15 @@ type UpdateHandler struct {
 	tokenQueries             *queries.CapabilityTokenQueries   // optional; delivers minted tokens to agents
 	orchestrator             LifecycleAdvancer                 // optional; auto-approve fast-path after a scan
 	reconciler               ReconcilerTrigger                 // optional; auto-bind tracked software after a scan
+	runner                   BackgroundRunner                  // optional; bounded fire-and-forget pool (SCALE-001 S2)
+}
+
+// BackgroundRunner is the bounded fire-and-forget pool (SCALE-001 S6/S2). The
+// report path used to launch raw `go func()` calls with no backpressure; routing
+// them through a runner caps steady-state concurrency and makes saturation
+// observable. Implemented by *taskrunner.Runner.
+type BackgroundRunner interface {
+	Go(name string, fn func())
 }
 
 // LifecycleAdvancer is the orchestrator hook the update handlers fire after a
@@ -109,6 +118,22 @@ func (h *UpdateHandler) SetReconciler(r ReconcilerTrigger) {
 	h.reconciler = r
 }
 
+// SetTaskRunner wires the bounded background pool. Nil falls back to launching a
+// plain goroutine per task (legacy unbounded behavior).
+func (h *UpdateHandler) SetTaskRunner(r BackgroundRunner) {
+	h.runner = r
+}
+
+// bg runs fn on the bounded pool when a runner is wired, else as a plain
+// goroutine. Keeps the report path's fire-and-forget work bounded and named.
+func (h *UpdateHandler) bg(name string, fn func()) {
+	if h.runner != nil {
+		h.runner.Go(name, fn)
+		return
+	}
+	go fn()
+}
+
 // EnqueueDryRun creates and signs the dry_run_update command for a package and
 // advances it to checking_dependencies, queuing a short heartbeat so the agent
 // polls promptly. Shared by the operator-triggered dry-run endpoint and the
@@ -272,7 +297,7 @@ func (h *UpdateHandler) ReportUpdates(c *gin.Context) {
 	// per server lifetime. Results (vulns or clean) are written back to the
 	// current_package_state metadata so the UI's "Known Vulnerabilities" card
 	// shows them before the operator approves.
-	go enqueueOSVChecks(events, h.updateQueries)
+	h.bg("osv_checks", func() { enqueueOSVChecks(events, h.updateQueries) })
 
 	// Fire the orchestrator's auto-approve fast-path so packages eligible by
 	// policy advance without waiting for the next timer sweep. Non-blocking and
@@ -283,7 +308,7 @@ func (h *UpdateHandler) ReportUpdates(c *gin.Context) {
 
 	// Fire the reconciler so newly-reported packages get automatic bindings.
 	if h.reconciler != nil {
-		go h.reconciler.ReconcileAll(context.Background())
+		h.bg("reconcile_all", func() { h.reconciler.ReconcileAll(context.Background()) })
 	}
 
 	// RECONCILE-001: scan-set closure (close-by-absence).
@@ -301,7 +326,7 @@ func (h *UpdateHandler) ReportUpdates(c *gin.Context) {
 		// Run closure async so it does not delay the HTTP response to the agent.
 		// Errors are logged; they do not fail the report (additive ingest already
 		// committed). The reconciler is not on the critical path.
-		go h.closeScanAbsentRows(agentID, req.Ecosystem, reportedSet)
+		h.bg("close_scan_absent", func() { h.closeScanAbsentRows(agentID, req.Ecosystem, reportedSet) })
 	}
 
 	c.JSON(http.StatusOK, gin.H{
@@ -1998,7 +2023,8 @@ func (h *UpdateHandler) ReportDependencies(c *gin.Context) {
 	// closure comes back clean, the orchestrator is triggered to auto-confirm if
 	// policy + window allow. Async so the agent's report returns promptly.
 	if update, err := h.updateQueries.GetUpdateByPackage(agentID, req.PackageType, req.PackageName); err == nil {
-		go h.checkClosureAndAdvance(update.ID, req.PackageType, req.Closure)
+		updateID, pkgType, closure := update.ID, req.PackageType, req.Closure
+		h.bg("closure_advance", func() { h.checkClosureAndAdvance(updateID, pkgType, closure) })
 	} else {
 		log.Printf("[ERROR] [server] [capability] closure_check_lookup_failed agent=%s pkg=%s/%s error=%v",
 			agentID, req.PackageType, req.PackageName, err)

server/internal/api/handlers/upstream.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"context"
 	"net/http"
 
 	"github.com/Fimeg/RedFlag/server/internal/database/queries"
@@ -16,12 +17,28 @@ type UpstreamHandler struct {
 	queries  *queries.UpstreamQueries
 	syncer   *upstream.Syncer
 	registry *upstream.Registry
+	runner   BackgroundRunner // optional; bounded fire-and-forget pool (SCALE-001 S2)
 }
 
 func NewUpstreamHandler(q *queries.UpstreamQueries, s *upstream.Syncer, r *upstream.Registry) *UpstreamHandler {
 	return &UpstreamHandler{queries: q, syncer: s, registry: r}
 }
 
+// SetTaskRunner wires the bounded background pool. Nil falls back to a plain
+// goroutine (legacy unbounded behavior).
+func (h *UpstreamHandler) SetTaskRunner(r BackgroundRunner) {
+	h.runner = r
+}
+
+// bg runs fn on the bounded pool when wired, else as a plain goroutine.
+func (h *UpstreamHandler) bg(name string, fn func()) {
+	if h.runner != nil {
+		h.runner.Go(name, fn)
+		return
+	}
+	go fn()
+}
+
 // List returns every tracked_software row. The dashboard panel filters
 // client-side; admin pages use the full list.
 func (h *UpstreamHandler) List(c *gin.Context) {
@@ -64,11 +81,15 @@ func (h *UpstreamHandler) Create(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
-	// Kick off an immediate sync so the new row gets populated without
-	// waiting a full tick. Fire-and-forget; the result writes to DB.
-	go func() {
-		_ = h.syncer.SyncOne(c.Request.Context(), row.ID)
-	}()
+	// Kick off an immediate sync so the new row gets populated without waiting a
+	// full tick. Fire-and-forget; the result writes to DB. Uses a background
+	// context, not the request's — the handler returns immediately, and the
+	// request context would be cancelled out from under the sync (the bug that
+	// silently killed the immediate sync's HTTP fetch).
+	id := row.ID
+	h.bg("upstream_sync_one", func() {
+		_ = h.syncer.SyncOne(context.Background(), id)
+	})
 	c.JSON(http.StatusCreated, row)
 }
 

server/internal/database/db.go

@@ -6,12 +6,24 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/jmoiron/sqlx"
 	_ "github.com/lib/pq"
 )
 
+// Connection-pool defaults. Sized for a single well-tuned server fronting 50–200
+// agents (SCALE-001 S1). The old 25/5 ceiling starved around ~30 concurrent agents
+// once the scheduler workers, sweeps, syncer, reconciler, and request handlers all
+// contended for the pool. All three are overridable via env for smaller/larger hosts.
+const (
+	defaultMaxOpenConns        = 100
+	defaultMaxIdleConns        = 25
+	defaultConnMaxLifetimeMins = 30
+)
+
 // DB wraps the database connection
 type DB struct {
 	*sqlx.DB
@@ -24,18 +36,41 @@ func Connect(databaseURL string) (*DB, error) {
 		return nil, fmt.Errorf("failed to connect to database: %w", err)
 	}
 
-	// Configure connection pool
-	db.SetMaxOpenConns(25)
-	db.SetMaxIdleConns(5)
+	// Configure connection pool (SCALE-001 S1). ConnMaxLifetime caps how long a
+	// connection is reused so the pool recycles cleanly behind poolers/restarts
+	// instead of holding stale handles indefinitely.
+	maxOpen := envInt("REDFLAG_DB_MAX_OPEN_CONNS", defaultMaxOpenConns)
+	maxIdle := envInt("REDFLAG_DB_MAX_IDLE_CONNS", defaultMaxIdleConns)
+	lifetimeMins := envInt("REDFLAG_DB_CONN_MAX_LIFETIME_MINUTES", defaultConnMaxLifetimeMins)
+	db.SetMaxOpenConns(maxOpen)
+	db.SetMaxIdleConns(maxIdle)
+	db.SetConnMaxLifetime(time.Duration(lifetimeMins) * time.Minute)
 
 	// Test the connection
 	if err := db.Ping(); err != nil {
 		return nil, fmt.Errorf("failed to ping database: %w", err)
 	}
 
+	log.Printf("[INFO] [server] [database] pool_configured max_open=%d max_idle=%d conn_max_lifetime_min=%d", maxOpen, maxIdle, lifetimeMins)
 	return &DB{db}, nil
 }
 
+// envInt reads a positive integer from env, falling back to def when unset, empty,
+// unparseable, or non-positive. A non-positive override is ignored so a stray
+// "0"/"-1" can't silently disable the pool ceiling.
+func envInt(key string, def int) int {
+	v := os.Getenv(key)
+	if v == "" {
+		return def
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil || n <= 0 {
+		log.Printf("[WARN] [server] [database] invalid_env key=%s value=%q falling_back=%d", key, v, def)
+		return def
+	}
+	return n
+}
+
 // Migrate runs database migrations with proper tracking and safety
 func (db *DB) Migrate(migrationsPath string) error {
 	// Create migrations table if it doesn't exist

server/internal/database/db_test.go

@@ -0,0 +1,31 @@
+package database
+
+import "testing"
+
+func TestEnvInt(t *testing.T) {
+	cases := []struct {
+		name string
+		set  bool
+		val  string
+		def  int
+		want int
+	}{
+		{"unset uses default", false, "", 100, 100},
+		{"valid override", true, "200", 100, 200},
+		{"empty uses default", true, "", 100, 100},
+		{"non-numeric uses default", true, "lots", 100, 100},
+		{"zero ignored (cannot disable ceiling)", true, "0", 100, 100},
+		{"negative ignored", true, "-5", 100, 100},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			const key = "REDFLAG_DB_TEST_ENVINT"
+			if c.set {
+				t.Setenv(key, c.val)
+			}
+			if got := envInt(key, c.def); got != c.want {
+				t.Errorf("envInt(%q, %d) = %d, want %d", c.val, c.def, got, c.want)
+			}
+		})
+	}
+}

server/internal/scheduler/scheduler.go

@@ -34,6 +34,13 @@ type Config struct {
 
 	// RateLimitPerSecond is max commands created per second (0 = unlimited)
 	RateLimitPerSecond int
+
+	// MaxDispatchPerTick caps how many due jobs are popped and dispatched in a
+	// single tick (SCALE-001 S4). Without a cap, a fleet whose subsystem jobs
+	// phase-align can dump thousands of jobs into one tick, a thundering herd
+	// against the DB pool. The remainder stays queued for the next tick. 0 =
+	// unlimited (legacy behavior).
+	MaxDispatchPerTick int
 }
 
 // DefaultConfig returns default configuration values
@@ -45,6 +52,7 @@ func DefaultConfig() Config {
 		NumWorkers:            10,
 		BackpressureThreshold: 5,
 		RateLimitPerSecond:    100,
+		MaxDispatchPerTick:    200,
 	}
 }
 
@@ -293,9 +301,15 @@ func (s *Scheduler) mainLoop() {
 func (s *Scheduler) processQueue() {
 	start := time.Now()
 
-	// Get all jobs due within lookahead window
+	// Get jobs due within the lookahead window, capped per tick (SCALE-001 S4)
+	// so an aligned fleet can't dump the whole queue into one tick. Overflow
+	// stays queued and is picked up next tick.
 	cutoff := time.Now().UTC().Add(s.config.LookaheadWindow)
-	dueJobs := s.queue.PopBefore(cutoff, 0) // No limit, get all
+	dueJobs := s.queue.PopBefore(cutoff, s.config.MaxDispatchPerTick)
+	if s.config.MaxDispatchPerTick > 0 && len(dueJobs) == s.config.MaxDispatchPerTick && s.queue.Len() > 0 {
+		log.Printf("[Scheduler] Dispatch cap hit: %d jobs this tick, %d still queued",
+			len(dueJobs), s.queue.Len())
+	}
 
 	if len(dueJobs) == 0 {
 		// No jobs due, just update stats

server/internal/services/reconciler.go

@@ -16,6 +16,12 @@ import (
 // Reconciler automatically matches agent-reported packages to tracked_software
 // entries using the cascade: Repology aliases -> container image patterns ->
 // exact source_ref name matching.
+//
+// mu + running together form a "single-flight" run-guard: mu protects the
+// running flag only. ReconcileAll acquires mu briefly to check/set running,
+// then releases it for the slow per-item work, then re-acquires to clear the
+// flag. This prevents overlapping ReconcileAll runs without holding the lock
+// across HTTP calls, DB writes, or the repology rate-limit sleep.
 type Reconciler struct {
 	agentSWQueries *queries.AgentTrackedSoftwareQueries
 	upstreamQ      *queries.UpstreamQueries
@@ -24,6 +30,7 @@ type Reconciler struct {
 	interval       time.Duration
 	shutdown       chan struct{}
 	mu             sync.Mutex
+	running        bool
 }
 
 func NewReconciler(
@@ -69,9 +76,24 @@ func (r *Reconciler) loop(ctx context.Context) {
 }
 
 // ReconcileAll runs the full reconciliation cascade across all tracked software.
+// It is single-flight: if a run is already in progress the new call returns
+// immediately without blocking. mu is held only for the brief flag check/set,
+// not across the slow per-item work (Repology HTTP + DB + rate-limit sleep).
 func (r *Reconciler) ReconcileAll(ctx context.Context) {
 	r.mu.Lock()
-	defer r.mu.Unlock()
+	if r.running {
+		r.mu.Unlock()
+		log.Printf("[INFO] [reconciler] reconcile_all skipped: already running")
+		return
+	}
+	r.running = true
+	r.mu.Unlock()
+
+	defer func() {
+		r.mu.Lock()
+		r.running = false
+		r.mu.Unlock()
+	}()
 
 	rows, err := r.upstreamQ.List()
 	if err != nil {
@@ -90,9 +112,10 @@ func (r *Reconciler) ReconcileAll(ctx context.Context) {
 }
 
 // ReconcileOne runs reconciliation for a single tracked_software entry.
+// It does not participate in the ReconcileAll run-guard: on-demand single-item
+// reconciliation is always allowed regardless of whether a background run is
+// in progress, and does not block the background run.
 func (r *Reconciler) ReconcileOne(ctx context.Context, sw models.TrackedSoftware) int {
-	r.mu.Lock()
-	defer r.mu.Unlock()
 	return r.reconcileOne(ctx, sw)
 }
 

server/internal/services/upstream/repology_cache.go

@@ -95,10 +95,22 @@ type repologyPackageEntry struct {
 
 // Refresh fetches the Repology /packages endpoint for a slug and upserts
 // the ecosystem-grouped alias rows into the cache.
+//
+// Lock analysis: rc.mu was previously held across the entire function. The
+// RepologyCache struct has no in-memory mutable state — queries and ttl are
+// both set at construction and never changed. GetAliases / GetAliasesByEcosystem
+// go straight to the database without touching mu. Therefore rc.mu guards
+// nothing that Refresh touches: the HTTP call, JSON decode, and DB upsert all
+// operate on purely local variables or the database. Holding the mutex across
+// those operations serialised concurrent Refresh calls for *different* slugs
+// against each other for no benefit. rc.mu is retained in the struct for
+// forward-compatibility should in-memory state be added later, but Refresh
+// no longer acquires it.
+//
+// Rate-limiting: the 500ms sleep is a politeness pause toward repology.org.
+// It is preserved, but moved outside any mutex so it does not block other
+// goroutines from reading cached state while we wait.
 func (rc *RepologyCache) Refresh(ctx context.Context, slug string) error {
-	rc.mu.Lock()
-	defer rc.mu.Unlock()
-
 	endpoint := fmt.Sprintf("https://repology.org/api/v1/project/%s/packages", url.PathEscape(strings.ToLower(slug)))
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
 	if err != nil {
@@ -148,6 +160,9 @@ func (rc *RepologyCache) Refresh(ctx context.Context, slug string) error {
 	}
 
 	log.Printf("[INFO] [upstream] [repology_cache] refreshed slug=%s ecosystems=%d entries=%d", slug, len(ecoPkgs), len(entries))
+	// Politeness rate-limit toward repology.org: 500ms between requests per
+	// caller. Kept outside any mutex so concurrent callers for different slugs
+	// do not block each other during the wait.
 	time.Sleep(500 * time.Millisecond)
 	return nil
 }

server/internal/services/upstream/syncer.go

@@ -3,6 +3,9 @@ package upstream
 import (
 	"context"
 	"log"
+	"os"
+	"strconv"
+	"sync"
 	"time"
 
 	"github.com/Fimeg/RedFlag/server/internal/database/queries"
@@ -10,14 +13,16 @@ import (
 	"github.com/gofrs/uuid/v5"
 )
 
+// defaultSyncWorkers is the concurrency cap used when
+// REDFLAG_UPSTREAM_SYNC_CONCURRENCY is unset, zero, or invalid.
+const defaultSyncWorkers = 5
+
 // Syncer walks tracked_software at a configurable interval, dispatches each
 // row to the appropriate ReleaseSource adapter, writes the result back, and
 // emits a drift event when latest_version moves.
 //
-// Scoping note: this is the v0.2.0.0 scaffold. The loop deliberately makes
-// one HTTP request per row sequentially. When the list grows past ~50 we
-// should pool, batch, and respect Repology's rate hints (Retry-After). For
-// now: simple, observable, correct.
+// tick() runs due rows concurrently up to syncWorkers goroutines (bounded by
+// a semaphore). Set REDFLAG_UPSTREAM_SYNC_CONCURRENCY to override the default.
 type Syncer struct {
 	queries      *queries.UpstreamQueries
 	registry     *Registry
@@ -26,6 +31,7 @@ type Syncer struct {
 	interval     time.Duration
 	staleness    time.Duration
 	batch        int
+	syncWorkers  int
 	shutdown     chan struct{}
 }
 
@@ -39,6 +45,14 @@ func NewSyncer(q *queries.UpstreamQueries, r *Registry, interval, staleness time
 	if batch <= 0 {
 		batch = 50
 	}
+	workers := defaultSyncWorkers
+	if raw := os.Getenv("REDFLAG_UPSTREAM_SYNC_CONCURRENCY"); raw != "" {
+		if n, err := strconv.Atoi(raw); err == nil && n > 0 {
+			workers = n
+		} else {
+			log.Printf("[WARN] [upstream] [syncer] invalid REDFLAG_UPSTREAM_SYNC_CONCURRENCY=%q, using default %d", raw, defaultSyncWorkers)
+		}
+	}
 	var ac *RepologyCache
 	if aliasQ != nil {
 		ac = NewRepologyCache(aliasQ)
@@ -51,6 +65,7 @@ func NewSyncer(q *queries.UpstreamQueries, r *Registry, interval, staleness time
 		interval:     interval,
 		staleness:    staleness,
 		batch:        batch,
+		syncWorkers:  workers,
 		shutdown:     make(chan struct{}),
 	}
 }
@@ -64,8 +79,8 @@ func (s *Syncer) Stop() {
 }
 
 func (s *Syncer) loop(ctx context.Context) {
-	log.Printf("[INFO] [upstream] [syncer] started interval=%s staleness=%s batch=%d sources=%v",
-		s.interval, s.staleness, s.batch, s.registry.Names())
+	log.Printf("[INFO] [upstream] [syncer] started interval=%s staleness=%s batch=%d workers=%d sources=%v",
+		s.interval, s.staleness, s.batch, s.syncWorkers, s.registry.Names())
 
 	// Run once immediately so a freshly-added row gets synced without
 	// waiting a full interval. Then settle into the cadence.
@@ -133,10 +148,25 @@ func (s *Syncer) tick(ctx context.Context) {
 	if len(rows) == 0 {
 		return
 	}
-	log.Printf("[INFO] [upstream] [syncer] processing %d due rows", len(rows))
+	log.Printf("[INFO] [upstream] [syncer] processing %d due rows workers=%d", len(rows), s.syncWorkers)
+
+	// Bounded-concurrency fan-out: a buffered semaphore channel limits the
+	// number of concurrent syncOne goroutines to s.syncWorkers. The WaitGroup
+	// ensures tick() does not return until all launched goroutines finish, so
+	// the next tick never overlaps with an in-flight batch.
+	sem := make(chan struct{}, s.syncWorkers)
+	var wg sync.WaitGroup
 	for _, row := range rows {
-		s.syncOne(ctx, row)
+		row := row // capture loop variable
+		sem <- struct{}{}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			defer func() { <-sem }()
+			s.syncOne(ctx, row)
+		}()
 	}
+	wg.Wait()
 }
 
 // SyncOne performs an on-demand sync for a single row (called by the

server/internal/taskrunner/taskrunner.go

@@ -0,0 +1,225 @@
+// Package taskrunner provides a bounded, observable runner for the server's
+// background work (SCALE-001 S6). It serves two roles:
+//
+//  1. A bounded ad-hoc pool (Go) for the fire-and-forget work that HTTP
+//     handlers previously launched as raw `go func()` calls with no
+//     backpressure (SCALE-001 S2). Steady-state concurrency is capped at the
+//     worker count. Bursts past the queue are run detached but counted as
+//     overflow, so saturation shows up in the health snapshot instead of being
+//     either silently dropped or blocking the request handler.
+//
+//  2. A registry of named periodic tasks (Every) with per-task jitter and a
+//     unified lifecycle, so background tickers become observable as a group and
+//     stop phase-aligning into DB thundering herds (SCALE-001 S7 groundwork).
+//
+// Every task run is panic-isolated: a panicking background job is recovered and
+// counted, never crashing the server.
+package taskrunner
+
+import (
+	"log"
+	"math/rand"
+	"runtime/debug"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Runner is a bounded background-work executor. The zero value is not usable;
+// construct one with New.
+type Runner struct {
+	workers int
+	queue   chan task
+	stop    chan struct{}
+	wg      sync.WaitGroup
+
+	// counters (atomic)
+	submitted uint64
+	completed uint64
+	overflow  uint64
+	panicked  uint64
+	inflight  int64
+
+	mu       sync.Mutex
+	periodic []*periodicTask
+	stopped  bool
+}
+
+type task struct {
+	name string
+	fn   func()
+}
+
+type periodicTask struct {
+	name     string
+	interval time.Duration
+	jitter   time.Duration
+	runs     uint64 // atomic
+	lastRun  int64  // atomic, unix nanos
+}
+
+// New starts a Runner with the given worker count and ad-hoc queue depth.
+// Non-positive values fall back to safe defaults.
+func New(workers, queueSize int) *Runner {
+	if workers <= 0 {
+		workers = 8
+	}
+	if queueSize <= 0 {
+		queueSize = 256
+	}
+	r := &Runner{
+		workers: workers,
+		queue:   make(chan task, queueSize),
+		stop:    make(chan struct{}),
+	}
+	for i := 0; i < workers; i++ {
+		r.wg.Add(1)
+		go r.worker()
+	}
+	log.Printf("[INFO] [server] [taskrunner] started workers=%d queue=%d", workers, queueSize)
+	return r
+}
+
+func (r *Runner) worker() {
+	defer r.wg.Done()
+	for {
+		select {
+		case <-r.stop:
+			return
+		case t := <-r.queue:
+			r.run(t)
+		}
+	}
+}
+
+// run executes a task with panic isolation and in-flight accounting.
+func (r *Runner) run(t task) {
+	atomic.AddInt64(&r.inflight, 1)
+	defer atomic.AddInt64(&r.inflight, -1)
+	defer func() {
+		if rec := recover(); rec != nil {
+			atomic.AddUint64(&r.panicked, 1)
+			log.Printf("[ERROR] [server] [taskrunner] task_panic name=%s recovered=%v\n%s", t.name, rec, debug.Stack())
+		}
+	}()
+	t.fn()
+	atomic.AddUint64(&r.completed, 1)
+}
+
+// Go schedules fn on the bounded pool. Under steady load concurrency is capped
+// at the worker count. If the queue is full (sustained burst) fn is run in a
+// detached goroutine and counted as overflow so the pressure is visible via
+// Snapshot rather than silently dropped or blocking the caller.
+func (r *Runner) Go(name string, fn func()) {
+	atomic.AddUint64(&r.submitted, 1)
+	t := task{name: name, fn: fn}
+	select {
+	case r.queue <- t:
+	default:
+		atomic.AddUint64(&r.overflow, 1)
+		log.Printf("[WARN] [server] [taskrunner] queue_saturated name=%s running_detached qlen=%d", name, len(r.queue))
+		go r.run(t)
+	}
+}
+
+// Every registers a named periodic task that runs fn on the given interval with
+// up to jitter random delay added per tick. Jitter decorrelates tickers so they
+// do not phase-align into DB thundering herds. The first run happens after one
+// interval (plus jitter). Registered tasks are reported by Snapshot and stopped
+// by Stop.
+func (r *Runner) Every(name string, interval, jitter time.Duration, fn func()) {
+	pt := &periodicTask{name: name, interval: interval, jitter: jitter}
+	r.mu.Lock()
+	r.periodic = append(r.periodic, pt)
+	r.mu.Unlock()
+
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-r.stop:
+				return
+			case <-ticker.C:
+				if jitter > 0 {
+					select {
+					case <-time.After(time.Duration(rand.Int63n(int64(jitter)))):
+					case <-r.stop:
+						return
+					}
+				}
+				atomic.AddUint64(&pt.runs, 1)
+				atomic.StoreInt64(&pt.lastRun, time.Now().UnixNano())
+				r.run(task{name: name, fn: fn})
+			}
+		}
+	}()
+}
+
+// Stop signals all workers and periodic tasks to exit and waits for them.
+// Idempotent.
+func (r *Runner) Stop() {
+	r.mu.Lock()
+	if r.stopped {
+		r.mu.Unlock()
+		return
+	}
+	r.stopped = true
+	r.mu.Unlock()
+	close(r.stop)
+	r.wg.Wait()
+}
+
+// Snapshot is a point-in-time view of the runner for health reporting.
+type Snapshot struct {
+	Workers   int                `json:"workers"`
+	QueueLen  int                `json:"queue_len"`
+	QueueCap  int                `json:"queue_cap"`
+	Inflight  int64              `json:"inflight"`
+	Submitted uint64             `json:"submitted"`
+	Completed uint64             `json:"completed"`
+	Overflow  uint64             `json:"overflow"`
+	Panicked  uint64             `json:"panicked"`
+	Periodic  []PeriodicSnapshot `json:"periodic"`
+}
+
+// PeriodicSnapshot reports one registered periodic task.
+type PeriodicSnapshot struct {
+	Name     string `json:"name"`
+	Interval string `json:"interval"`
+	Jitter   string `json:"jitter"`
+	Runs     uint64 `json:"runs"`
+	LastRun  string `json:"last_run,omitempty"`
+}
+
+// Snapshot returns the current counters and registered periodic tasks.
+func (r *Runner) Snapshot() Snapshot {
+	r.mu.Lock()
+	periodic := make([]PeriodicSnapshot, 0, len(r.periodic))
+	for _, pt := range r.periodic {
+		ps := PeriodicSnapshot{
+			Name:     pt.name,
+			Interval: pt.interval.String(),
+			Jitter:   pt.jitter.String(),
+			Runs:     atomic.LoadUint64(&pt.runs),
+		}
+		if ln := atomic.LoadInt64(&pt.lastRun); ln > 0 {
+			ps.LastRun = time.Unix(0, ln).UTC().Format(time.RFC3339)
+		}
+		periodic = append(periodic, ps)
+	}
+	r.mu.Unlock()
+	return Snapshot{
+		Workers:   r.workers,
+		QueueLen:  len(r.queue),
+		QueueCap:  cap(r.queue),
+		Inflight:  atomic.LoadInt64(&r.inflight),
+		Submitted: atomic.LoadUint64(&r.submitted),
+		Completed: atomic.LoadUint64(&r.completed),
+		Overflow:  atomic.LoadUint64(&r.overflow),
+		Panicked:  atomic.LoadUint64(&r.panicked),
+		Periodic:  periodic,
+	}
+}

server/internal/taskrunner/taskrunner_test.go

@@ -0,0 +1,132 @@
+package taskrunner
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestGo_RunsAllSubmittedTasks(t *testing.T) {
+	r := New(4, 64)
+	defer r.Stop()
+
+	const n = 100
+	var got int64
+	var wg sync.WaitGroup
+	wg.Add(n)
+	for i := 0; i < n; i++ {
+		r.Go("unit", func() {
+			atomic.AddInt64(&got, 1)
+			wg.Done()
+		})
+	}
+	wg.Wait()
+
+	if got != n {
+		t.Fatalf("ran %d tasks, want %d", got, n)
+	}
+	s := r.Snapshot()
+	if s.Submitted != n {
+		t.Errorf("submitted=%d, want %d", s.Submitted, n)
+	}
+	if s.Completed != n {
+		t.Errorf("completed=%d, want %d", s.Completed, n)
+	}
+}
+
+func TestRun_PanicIsIsolatedAndCounted(t *testing.T) {
+	r := New(2, 8)
+	defer r.Stop()
+
+	var done sync.WaitGroup
+	done.Add(2)
+	r.Go("boom", func() { defer done.Done(); panic("kaboom") })
+	r.Go("ok", func() { defer done.Done() })
+	done.Wait()
+
+	// Give the panicking task's deferred counters a moment to settle.
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if r.Snapshot().Panicked >= 1 {
+			break
+		}
+		time.Sleep(2 * time.Millisecond)
+	}
+	if p := r.Snapshot().Panicked; p != 1 {
+		t.Errorf("panicked=%d, want 1", p)
+	}
+}
+
+func TestGo_OverflowRunsDetachedAndIsCounted(t *testing.T) {
+	// One worker, queue of one, blocked work forces overflow on later submits.
+	r := New(1, 1)
+	defer r.Stop()
+
+	release := make(chan struct{})
+	var ran int64
+	var wg sync.WaitGroup
+
+	// Occupy the single worker.
+	wg.Add(1)
+	r.Go("blocker", func() {
+		defer wg.Done()
+		atomic.AddInt64(&ran, 1)
+		<-release
+	})
+	// Let the worker pick up the blocker so it's no longer in the queue.
+	time.Sleep(20 * time.Millisecond)
+
+	// Submit more than the queue can hold; the excess must overflow (detached),
+	// never be dropped.
+	const extra = 20
+	wg.Add(extra)
+	for i := 0; i < extra; i++ {
+		r.Go("extra", func() {
+			defer wg.Done()
+			atomic.AddInt64(&ran, 1)
+		})
+	}
+
+	close(release)
+	wg.Wait()
+
+	if ran != extra+1 {
+		t.Fatalf("ran %d tasks, want %d (no work lost)", ran, extra+1)
+	}
+	if r.Snapshot().Overflow == 0 {
+		t.Error("expected overflow > 0 under saturation")
+	}
+}
+
+func TestEvery_RunsAndIsReported(t *testing.T) {
+	r := New(2, 8)
+	defer r.Stop()
+
+	var hits int64
+	r.Every("tick", 10*time.Millisecond, 0, func() { atomic.AddInt64(&hits, 1) })
+
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if atomic.LoadInt64(&hits) >= 2 {
+			break
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	if atomic.LoadInt64(&hits) < 2 {
+		t.Fatalf("periodic task ran %d times, want >= 2", hits)
+	}
+	s := r.Snapshot()
+	if len(s.Periodic) != 1 || s.Periodic[0].Name != "tick" {
+		t.Fatalf("periodic snapshot = %+v, want one task named tick", s.Periodic)
+	}
+	if s.Periodic[0].Runs < 2 {
+		t.Errorf("periodic runs=%d, want >= 2", s.Periodic[0].Runs)
+	}
+}
+
+func TestStop_IsIdempotent(t *testing.T) {
+	r := New(2, 4)
+	r.Stop()
+	r.Stop() // must not panic on double close
+}