qa: add retransmit metro test to qa.testnet (#4156)

## Summary of Changes
* This PR reworks the `TestQA_MulticastSettlement`.
* Deleted `e2e/qa_shred_settlement_test.go.` since it only contained one
setup method that I moved into the `TestQA_MulticastSettlement` file.
* It checks the env we are in, if mainnet, continue as normal, if
testnet and program config retransmit flag off, continue as normal too.
If testnet and program config retransmit flag on do the following:

1. Assert that when trying to buy a seat in a metro other than frankfurt
fails
2. Try to buy a seat in frankfurt and assert that the user accesspass
has access to these multicast groups: `edge-solana-retrans`

* The PR deletes `e2e/qa_retransmit_only_settlement_test.go`.
`TestQA_RetransmitOnlySettlement` drove the same positive path, so the
reworked test covers it now. Its device selector, its metro-code
formatter and its group assertion moved into the multicast test file.
* The group codes now come from one flag, `-retransmit-group-codes`,
which replaces `-leader-group-code` and `-retransmit-group-code`. The
check requires the seat's groups to equal that list, so any extra group
fails it.
* `sdk/shreds/go/state.go` gains `IsRetransmitOnlyOnboardingEnforced`,
which reads bit 7 of the program config flags.
* `e2e/internal/qa/client_settlement.go` gains that same read, plus
`ClosestNonRetransmitOnlyDevice` for the device the rejection step pays
on.
* A companion change in `malbeclabs/infra` passes
`-retransmit-group-codes=edge-solana-retrans` in `qa.testnet.yml` and
deletes `qa.retransmits.testnet.yml`. Merge both together, or the hourly
job runs the group check without the codes it needs.

Author: martinsander00

CHANGELOG.md

@@ -23,6 +23,8 @@ All notable changes to this project will be documented in this file.
   - Samples dropped because a partition's onchain account is full are now counted too, under `reason="account_full"` plus `submitter_account_full` on the errors counter. That path reports success to its caller, so a warning was the only trace, and its count was wrong: it reported the whole flushed partition rather than the samples actually lost. (#4145)
   - A failed submission now retries from the first unwritten sample rather than restarting at the beginning of the flushed partition, and only the unwritten remainder is requeued. Previously a mid-partition error made every subsequent attempt re-send batches that were already onchain, appending those samples a second time and pushing the account toward the sample cap it is measured against. (#4145)
   - Agent logs now identify the ledger RPC endpoint in use, and the peer count and the stale-program-data warning report state transitions instead of firing on every refresh. New lines name the resolved remote address of each connection (a bad load balancer address behind a hostname was invisible before), and a refresh that finds no peers is now a warning rather than a Debug line. The stale-cache warning also moves off the package-global `slog` onto the agent's own logger, so it is formatted and leveled with everything else. New: `doublezero_device_telemetry_agent_peers` gauge, and `pinger_epoch_fetch` on the errors counter for every exhausted epoch fetch. (#4147)
+- QA
+  - Rework existing TestQA_MulticastSettlement and adapt it to the new `FLAG_RETRANSMIT_ONLY_ONBOARDING_ENFORCED_BIT` flag. Test now checks for this flag in the ProgramConfig solana account and depending on if it's on or off tries to assert that no new user can subscribe to a non retransmit-only metro unless that metro has the retransmit-only flag enabled in the MetroHistory account. (#4156)
 
 ## [v0.33.0](https://github.com/malbeclabs/doublezero/compare/client/v0.32.0...client/v0.33.0) - 2026-07-31
 

e2e/internal/qa/client.go

@@ -507,6 +507,21 @@ func (c *Client) GetServiceabilityUser(ctx context.Context) (*serviceability.Use
 	return nil, fmt.Errorf("serviceability user not found for client IP %s on host %s", publicIP, c.Host)
 }
 
+func (c *Client) GetMulticastServiceabilityUser(ctx context.Context) (*serviceability.User, error) {
+	data, err := getProgramDataWithRetry(ctx, c.serviceability)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get program data on host %s: %w", c.Host, err)
+	}
+	publicIP := c.publicIP.To4().String()
+	for i := range data.Users {
+		user := &data.Users[i]
+		if net.IP(user.ClientIp[:]).String() == publicIP && user.UserType == serviceability.UserTypeMulticast {
+			return user, nil
+		}
+	}
+	return nil, fmt.Errorf("multicast serviceability user not found for client IP %s on host %s", publicIP, c.Host)
+}
+
 func (c *Client) GetOwnerPubkey(ctx context.Context) (solana.PublicKey, error) {
 	user, err := c.GetServiceabilityUser(ctx)
 	if err != nil {

e2e/internal/qa/client_multicast.go

@@ -141,6 +141,20 @@ func (c *Client) GetMulticastGroup(ctx context.Context, code string) (*Multicast
 	return nil, nil
 }
 
+// MulticastGroupCodes maps every multicast group pubkey to its code, so a caller
+// holding pubkeys off a user account can name them in a log or a failure.
+func (c *Client) MulticastGroupCodes(ctx context.Context) (map[solana.PublicKey]string, error) {
+	data, err := getProgramDataWithRetry(ctx, c.serviceability)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get program data on host %s: %w", c.Host, err)
+	}
+	codes := make(map[solana.PublicKey]string, len(data.MulticastGroups))
+	for _, group := range data.MulticastGroups {
+		codes[solana.PublicKeyFromBytes(group.PubKey[:])] = group.Code
+	}
+	return codes, nil
+}
+
 func (c *Client) CreateMulticastGroup(ctx context.Context, code string, maxBandwidth string) (*MulticastGroup, error) {
 	c.log.Debug("Creating multicast group", "host", c.Host, "code", code, "maxBandwidth", maxBandwidth)
 	resp, err := c.grpcClient.CreateMulticastGroup(ctx, &pb.CreateMulticastGroupRequest{

e2e/internal/qa/client_settlement.go

@@ -164,9 +164,30 @@ func (c *Client) ClosestRetransmitOnlyDevice(ctx context.Context) (*Device, map[
 		return nil, retransmitOnly, nil
 	}
 
+	device, err := c.closestDeviceInMetros(ctx, retransmitOnly, true)
+	if err != nil {
+		return nil, retransmitOnly, err
+	}
+	return device, retransmitOnly, nil
+}
+
+// ClosestNonRetransmitOnlyDevice returns the reachable device with the lowest
+// average latency whose metro is not flagged retransmit-only. A nil device means
+// every reachable metro is flagged, so no metro is left to reject a new seat.
+func (c *Client) ClosestNonRetransmitOnlyDevice(ctx context.Context) (*Device, error) {
+	retransmitOnly, err := c.RetransmitOnlyExchangeKeys(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return c.closestDeviceInMetros(ctx, retransmitOnly, false)
+}
+
+// closestDeviceInMetros returns the lowest-latency reachable device whose metro
+// membership in exchangeKeys equals want.
+func (c *Client) closestDeviceInMetros(ctx context.Context, exchangeKeys map[string]bool, want bool) (*Device, error) {
 	latencies, err := c.GetLatency(ctx)
 	if err != nil {
-		return nil, retransmitOnly, fmt.Errorf("failed to get latency on host %s: %w", c.Host, err)
+		return nil, fmt.Errorf("failed to get latency on host %s: %w", c.Host, err)
 	}
 
 	var bestDevice *Device
@@ -176,7 +197,7 @@ func (c *Client) ClosestRetransmitOnlyDevice(ctx context.Context) (*Device, map[
 			continue
 		}
 		device, ok := c.devices[l.DeviceCode]
-		if !ok || !retransmitOnly[device.ExchangePubKey] {
+		if !ok || exchangeKeys[device.ExchangePubKey] != want {
 			continue
 		}
 		if l.AvgLatencyNs < bestAvg {
@@ -185,9 +206,10 @@ func (c *Client) ClosestRetransmitOnlyDevice(ctx context.Context) (*Device, map[
 		}
 	}
 	if bestDevice != nil {
-		c.log.Debug("Determined closest retransmit-only device", "host", c.Host, "deviceCode", bestDevice.Code, "avgLatencyNs", bestAvg)
+		c.log.Debug("Determined closest device", "host", c.Host, "deviceCode", bestDevice.Code,
+			"avgLatencyNs", bestAvg, "retransmitOnly", want)
 	}
-	return bestDevice, retransmitOnly, nil
+	return bestDevice, nil
 }
 
 // FeedSeatPrice calls the FeedSeatPrice RPC to query seat pricing for a single
@@ -769,6 +791,19 @@ func (c *Client) IsSeatProratingEnabled(ctx context.Context) (bool, error) {
 	return cfg.IsProratedServiceEnabled(), nil
 }
 
+func (c *Client) IsRetransmitOnlyOnboardingEnforced(ctx context.Context) (bool, error) {
+	programID, err := solana.PublicKeyFromBase58(c.ShredSubscriptionProgramID)
+	if err != nil {
+		return false, fmt.Errorf("failed to parse shred subscription program ID %q: %w", c.ShredSubscriptionProgramID, err)
+	}
+
+	cfg, err := c.shredsClient(programID).FetchProgramConfig(ctx)
+	if err != nil {
+		return false, fmt.Errorf("failed to fetch program config on host %s: %w", c.Host, err)
+	}
+	return cfg.IsRetransmitOnlyOnboardingEnforced(), nil
+}
+
 // IsProgramPaused returns true if the shred-subscription program config has
 // the paused flag set. While paused, the oracle cannot ack instant seat
 // allocation requests, which leaves the seat un-withdrawable.