qa: add retransmit metro test to qa.testnet

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.