0.2.3.0: fix OSV supply-chain checks — bounded concurrency, persist-driven dedup

- Replace unbounded goroutine fan-out with bounded pool (8 concurrent)
  so 300-package dnf scans no longer timeout every request against api.osv.dev
- Drop in-memory osvDedup sync.Map; gate on persisted supply_chain_checked_at
  so the dedup survives restart and failed checks retry naturally
- On query failure, record the error without checked_at so the package stays
  a candidate for the next cycle (ETHOS: errors are history, assume failure)
- Shared RunOSVChecks in services/ used by both scan path and startup backfill
- Add FreshSupplyChainPackages query for persist-driven freshness lookup
- Bump version to 0.2.3.0

README.md

@@ -234,7 +234,7 @@ I am a Systems Architect with 25 years on the frontier. I build sovereign agent
 - **Consulting Expeditions ($175/hr):** Hands-on architecture, network engineering, and system hardening.
 - **Incident Response ($250/hr):** Ransomware restoration, AD/DNS rebuilds, and pulling your servers back from the void.
 
-**Contact:** casey@samaritansolutions.net | [LinkedIn](https://www.linkedin.com/in/casey-tunturi) | [GitHub Sponsors](https://github.com/sponsors/Fimeg)
+**Contact:** casey@samaritansolutions.net | [LinkedIn](https://www.linkedin.com/in/casey-tunturi) | [GitHub Sponsors](https://github.com/sponsors/Fimeg) | [Discord](https://discord.gg/TReG3mZC4Y)
 
 ---
 

docker-compose.yml

@@ -21,7 +21,7 @@ services:
       context: .
       dockerfile: ./server/Dockerfile
       args:
-        BUILD_VERSION: ${BUILD_VERSION:-0.2.2.0}
+        BUILD_VERSION: ${BUILD_VERSION:-0.2.3.0}
     container_name: redflag-server
     volumes:
       - server-config:/app/config

server/cmd/server/main.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"crypto/ed25519"
 	"encoding/hex"
-	"encoding/json"
 	"flag"
 	"fmt"
 	"log"
@@ -13,7 +12,9 @@ import (
 	"time"
 
 	"github.com/Fimeg/RedFlag/server/internal/api/handlers"
+	"github.com/Fimeg/RedFlag/server/internal/models"
 	"github.com/Fimeg/RedFlag/server/internal/api/middleware"
+	"github.com/google/uuid"
 	"github.com/Fimeg/RedFlag/server/internal/command"
 	"github.com/Fimeg/RedFlag/server/internal/config"
 	"github.com/Fimeg/RedFlag/server/internal/database"
@@ -904,36 +905,24 @@ func backfillOSVChecks(updateQueries *queries.UpdateQueries) {
 		log.Printf("[WARNING] [server] [supply_chain] backfill_query_failed error=%v", err)
 		return
 	}
-	log.Printf("[INFO] [server] [supply_chain] backfill_candidates count=%d", len(rows))
+
+	var reqs []services.OSVCheckRequest
 	for _, r := range rows {
 		if !services.NeedsSupplyChainCheck(r.PackageType) {
 			continue
 		}
-		pkgType, pkgName, version, agentID := r.PackageType, r.PackageName, r.Version, r.AgentID
-		go func() {
-			ecosystem := services.EcosystemFromPackageType(pkgType)
-			result := services.CheckOSVVulnerabilities(pkgName, ecosystem, version)
-			if result == nil {
-				return
-			}
-			meta := map[string]interface{}{
-				"supply_chain_checked_at": result.CheckedAt.UTC().Format(time.RFC3339),
-			}
-			if len(result.Vulnerabilities) > 0 {
-				vulnJSON, err := json.Marshal(result.Vulnerabilities)
-				if err != nil {
-					log.Printf("[WARNING] [supply_chain] backfill_marshal_failed pkg=%s error=%v", pkgName, err)
-					return
-				}
-				meta["supply_chain_vulns"] = string(vulnJSON)
-				log.Printf("[SECURITY] [supply_chain] backfill_vulns_found pkg=%s type=%s ecosystem=%s count=%d",
-					pkgName, pkgType, ecosystem, len(result.Vulnerabilities))
-			}
-			if err := updateQueries.StoreSupplyChainMetadata(agentID, pkgType, pkgName, meta); err != nil {
-				log.Printf("[WARNING] [supply_chain] backfill_store_failed pkg=%s error=%v", pkgName, err)
-			}
-		}()
+		reqs = append(reqs, services.OSVCheckRequest{
+			AgentID: r.AgentID,
+			PkgType: r.PackageType,
+			PkgName: r.PackageName,
+			Version: r.Version,
+		})
 	}
-	log.Printf("[INFO] [server] [supply_chain] backfill_enqueued")
+	log.Printf("[INFO] [server] [supply_chain] backfill_candidates count=%d", len(reqs))
+
+	services.RunOSVChecks(reqs, func(agentID uuid.UUID, pkgType, pkgName string, meta map[string]interface{}) error {
+		return updateQueries.StoreSupplyChainMetadata(agentID, pkgType, pkgName, models.JSONB(meta))
+	})
+	log.Printf("[INFO] [server] [supply_chain] backfill_complete count=%d", len(reqs))
 }
 

server/internal/api/handlers/updates.go

@@ -12,7 +12,6 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/Fimeg/RedFlag/server/internal/capability"
@@ -24,10 +23,6 @@ import (
 	"github.com/google/uuid"
 )
 
-// osvDedup avoids re-querying OSV.dev for the same package+version across
-// multiple scan cycles. Key format: "pkgType:pkgName:version".
-var osvDedup sync.Map
-
 // isValidResult checks if the result value complies with the database constraint
 func isValidResult(result string) bool {
 	validResults := map[string]bool{
@@ -246,47 +241,59 @@ func (h *UpdateHandler) ReportUpdates(c *gin.Context) {
 	})
 }
 
-// enqueueOSVChecks fires an OSV.dev query for each unique package+version in the
-// batch. Already-checked combos are skipped via osvDedup. Results are stored in
-// current_package_state.metadata for UI visibility.
+// enqueueOSVChecks checks each unique package+version in the batch against
+// OSV.dev with bounded concurrency, persisting every outcome (clean, vuln, or
+// recorded failure) to current_package_state.metadata. Packages whose stored
+// result is still fresh (per OSVRecheckInterval) are skipped via the persisted
+// timestamp — not a process-local cache — so the dedup survives restart and a
+// failed check is retried next cycle.
 func enqueueOSVChecks(events []models.UpdateEvent, q *queries.UpdateQueries) {
+	type dedupKey struct {
+		agentID                 uuid.UUID
+		pkgType, pkgName, version string
+	}
+	seen := make(map[dedupKey]bool)
+	freshByAgent := make(map[uuid.UUID]map[string]bool)
+	var reqs []services.OSVCheckRequest
+
 	for _, e := range events {
 		if !services.NeedsSupplyChainCheck(e.PackageType) {
 			continue
 		}
-		key := fmt.Sprintf("%s:%s:%s", e.PackageType, e.PackageName, e.VersionTo)
-		if _, seen := osvDedup.LoadOrStore(key, true); seen {
+		k := dedupKey{e.AgentID, e.PackageType, e.PackageName, e.VersionTo}
+		if seen[k] {
 			continue
 		}
-		// Capture loop variables for the goroutine
-		pkgType, pkgName, version, agentID := e.PackageType, e.PackageName, e.VersionTo, e.AgentID
-		go func() {
-			ecosystem := services.EcosystemFromPackageType(pkgType)
-			result := services.CheckOSVVulnerabilities(pkgName, ecosystem, version)
-			if result == nil {
-				return
+		seen[k] = true
+
+		fresh, ok := freshByAgent[e.AgentID]
+		if !ok {
+			f, err := q.FreshSupplyChainPackages(e.AgentID, services.OSVRecheckInterval)
+			if err != nil {
+				log.Printf("[WARNING] [supply_chain] fresh_query_failed agent=%s error=%v", e.AgentID, err)
+				f = map[string]bool{}
 			}
-			meta := models.JSONB{
-				"supply_chain_checked_at": result.CheckedAt.UTC().Format(time.RFC3339),
-			}
-			if len(result.Vulnerabilities) > 0 {
-				vulnJSON, err := json.Marshal(result.Vulnerabilities)
-				if err != nil {
-					log.Printf("[WARNING] [supply_chain] vuln_marshal_failed pkg=%s error=%v", pkgName, err)
-					return
-				}
-				meta["supply_chain_vulns"] = string(vulnJSON)
-				log.Printf("[SECURITY] [supply_chain] vulns_found pkg=%s type=%s ecosystem=%s count=%d",
-					pkgName, pkgType, ecosystem, len(result.Vulnerabilities))
-			} else {
-				log.Printf("[INFO] [supply_chain] clean pkg=%s type=%s ecosystem=%s version=%s",
-					pkgName, pkgType, ecosystem, version)
-			}
-			if err := q.StoreSupplyChainMetadata(agentID, pkgType, pkgName, meta); err != nil {
-				log.Printf("[WARNING] [supply_chain] metadata_store_failed pkg=%s error=%v", pkgName, err)
-			}
-		}()
+			fresh = f
+			freshByAgent[e.AgentID] = fresh
+		}
+		if fresh[e.PackageType+"\x00"+e.PackageName] {
+			continue
+		}
+
+		reqs = append(reqs, services.OSVCheckRequest{
+			AgentID: e.AgentID,
+			PkgType: e.PackageType,
+			PkgName: e.PackageName,
+			Version: e.VersionTo,
+		})
 	}
+
+	if len(reqs) == 0 {
+		return
+	}
+	services.RunOSVChecks(reqs, func(agentID uuid.UUID, pkgType, pkgName string, meta map[string]interface{}) error {
+		return q.StoreSupplyChainMetadata(agentID, pkgType, pkgName, models.JSONB(meta))
+	})
 }
 
 // ListUpdates retrieves updates with filtering using the new state table

server/internal/database/queries/updates.go

@@ -1317,6 +1317,33 @@ func (q *UpdateQueries) StoreSupplyChainMetadata(agentID uuid.UUID, pkgType, pkg
 	return err
 }
 
+// FreshSupplyChainPackages returns the set of (package_type, package_name) for
+// an agent whose stored supply-chain result is newer than within. Used by the
+// scan path to skip packages already checked recently. Keyed "type\x00name".
+func (q *UpdateQueries) FreshSupplyChainPackages(agentID uuid.UUID, within time.Duration) (map[string]bool, error) {
+	query := `
+		SELECT package_type, package_name
+		FROM current_package_state
+		WHERE agent_id = $1
+		  AND metadata ? 'supply_chain_checked_at'
+		  AND (metadata->>'supply_chain_checked_at')::timestamptz > NOW() - make_interval(secs => $2)`
+	rows, err := q.db.Query(query, agentID, within.Seconds())
+	if err != nil {
+		return nil, fmt.Errorf("query fresh supply-chain packages: %w", err)
+	}
+	defer rows.Close()
+
+	fresh := make(map[string]bool)
+	for rows.Next() {
+		var pkgType, pkgName string
+		if err := rows.Scan(&pkgType, &pkgName); err != nil {
+			return nil, fmt.Errorf("scan fresh supply-chain row: %w", err)
+		}
+		fresh[pkgType+"\x00"+pkgName] = true
+	}
+	return fresh, rows.Err()
+}
+
 // GetUncheckedPackages returns distinct (package_type, package_name, version_to,
 // agent_id) tuples from current_package_state that have never had a supply-chain
 // check. Used by the startup backfill.

server/internal/services/supply_chain.go

@@ -5,7 +5,10 @@ import (
 	"encoding/json"
 	"log"
 	"net/http"
+	"sync"
 	"time"
+
+	"github.com/google/uuid"
 )
 
 // OSVQueryRequest is sent to the OSV.dev API.
@@ -47,6 +50,84 @@ type VulnerabilityInfo struct {
 
 var osvHTTPClient = &http.Client{Timeout: 10 * time.Second}
 
+// OSVRecheckInterval is how long a stored supply-chain result is treated as
+// fresh. The scan path skips packages checked more recently than this so a
+// frequent scan cadence does not re-hit OSV.dev for every package every cycle;
+// the periodic recheck still picks up newly published advisories.
+const OSVRecheckInterval = 6 * time.Hour
+
+// osvConcurrency bounds simultaneous OSV.dev requests. A single query takes
+// ~1.5s; firing one goroutine per package (hundreds at once, as the old scan
+// and backfill paths did) saturated the shared client and every request hit
+// the 10s deadline together. Cap the fan-out.
+const osvConcurrency = 8
+
+// OSVCheckRequest is one package to check against OSV.dev.
+type OSVCheckRequest struct {
+	AgentID uuid.UUID
+	PkgType string
+	PkgName string
+	Version string
+}
+
+// OSVStoreFunc persists a supply-chain result (or a recorded failure) for one
+// package. Implemented by the caller against current_package_state so this
+// package keeps no database dependency.
+type OSVStoreFunc func(agentID uuid.UUID, pkgType, pkgName string, meta map[string]interface{}) error
+
+// RunOSVChecks queries OSV.dev for each request with bounded concurrency and
+// persists every outcome through store. A clean result records
+// supply_chain_checked_at; a hit also records supply_chain_vulns; a query
+// failure records supply_chain_check_error WITHOUT a checked_at timestamp, so
+// the package stays a candidate for the next run rather than being silently
+// dropped (ETHOS: errors are history, assume failure). Blocks until done.
+func RunOSVChecks(reqs []OSVCheckRequest, store OSVStoreFunc) {
+	sem := make(chan struct{}, osvConcurrency)
+	var wg sync.WaitGroup
+	for _, r := range reqs {
+		r := r
+		wg.Add(1)
+		sem <- struct{}{}
+		go func() {
+			defer wg.Done()
+			defer func() { <-sem }()
+
+			ecosystem := EcosystemFromPackageType(r.PkgType)
+			result := CheckOSVVulnerabilities(r.PkgName, ecosystem, r.Version)
+
+			meta := map[string]interface{}{}
+			if result == nil {
+				// Query failed; the underlying error is already logged in
+				// CheckOSVVulnerabilities. Record the failure and leave
+				// checked_at unset so this package is retried next cycle.
+				meta["supply_chain_check_error"] = "osv_query_failed"
+				meta["supply_chain_error_at"] = time.Now().UTC().Format(time.RFC3339)
+			} else {
+				meta["supply_chain_checked_at"] = result.CheckedAt.UTC().Format(time.RFC3339)
+				meta["supply_chain_check_error"] = nil // clear any prior failure
+				if len(result.Vulnerabilities) > 0 {
+					vulnJSON, err := json.Marshal(result.Vulnerabilities)
+					if err != nil {
+						log.Printf("[WARNING] [supply_chain] vuln_marshal_failed pkg=%s error=%v", r.PkgName, err)
+						return
+					}
+					meta["supply_chain_vulns"] = string(vulnJSON)
+					log.Printf("[SECURITY] [supply_chain] vulns_found pkg=%s type=%s ecosystem=%s count=%d",
+						r.PkgName, r.PkgType, ecosystem, len(result.Vulnerabilities))
+				} else {
+					log.Printf("[INFO] [supply_chain] clean pkg=%s type=%s ecosystem=%s version=%s",
+						r.PkgName, r.PkgType, ecosystem, r.Version)
+				}
+			}
+
+			if err := store(r.AgentID, r.PkgType, r.PkgName, meta); err != nil {
+				log.Printf("[WARNING] [supply_chain] metadata_store_failed pkg=%s error=%v", r.PkgName, err)
+			}
+		}()
+	}
+	wg.Wait()
+}
+
 // CheckOSVVulnerabilities queries the OSV.dev API for known vulnerabilities
 // matching the given package name, ecosystem, and version. Returns nil slice
 // with no error on clean results or API failure (fail-open).

server/internal/version/versions.go

@@ -12,8 +12,8 @@ import (
 
 // Build-time injected version information (SERVER AUTHORITY)
 var (
-	AgentVersion    = "0.2.2.0"
-	ConfigVersion   = "0.2.2.0"
+	AgentVersion    = "0.2.3.0"
+	ConfigVersion   = "0.2.3.0"
 	MinAgentVersion = "0.1.22"
 )