Preserve sim prewarm across transport rotation

Author: bitprecipice

connect/SIM-LATENCY.md

@@ -101,9 +101,10 @@ evidence (round-trip link latency and bytes/second), materializes
 final `client_connection_reliability_score` rows
 for every score lookback — rather than backfilling raw reliability blocks,
 which fought the running-window/shift/degraded maintenance. The performance
-evidence is attached to the current connection so later pipeline refreshes keep
-seeing it; connection tests belong to a particular platform transport, and a
-new active transport can otherwise make the entire fixture look untested. Each
+evidence is attached to the current connection, and prewarmed pipeline refreshes
+restore it on the provider-scoped derived snapshot; connection tests belong to a
+particular platform transport, and a new active transport can otherwise make
+the entire fixture look untested. Each
 reliability row uses the fixture provider's seeded
 `uptime / (uptime + downtime)` duty cycle, so a mature mobile churn profile does
 not rank as perfectly reliable. The pipeline then runs in prewarmed mode

connect/sim-latency/README.md

@@ -360,8 +360,8 @@ The warm-up phases and the knobs that make it **as fast as possible**:
 `--prewarm` writes the fixture's round-trip latency and bandwidth evidence onto
 each active connection, then writes the final reliability scores for every
 connected provider using its seeded uptime duty cycle, rather than replaying
-~8.4h of history. Attaching the evidence to the active connection makes later
-pipeline snapshots preserve it; tests belong to one transport, so a replacement
+~8.4h of history. Prewarmed pipeline refreshes restore that evidence on the
+provider-scoped snapshot; tests belong to one transport, so a replacement
 transport can otherwise make an established provider look untested.
 `--prewarm 0` restores
 the true cold start and live connection-test behavior. A shorter

connect/sim-latency/provision.go

@@ -301,9 +301,15 @@ func generatedClientPoolEntries(config *Config) []ProviderEntry {
 //
 // The pipeline must run in prewarmed mode afterwards (Services.SetPrewarmed),
 // so the periodic reliability-score recompute does not overwrite these rows;
-// it keeps refreshing the location reliabilities (so churn still gates
-// selection) and re-exporting the redis samples.
-func provisionPrewarm(ctx context.Context, lookback time.Duration, entries []ProviderEntry) error {
+// it refreshes connected/location state, restores provider-scoped fixture
+// performance after mechanical transport replacement, and re-exports the
+// redis samples (so churn still gates selection).
+func provisionPrewarm(
+	ctx context.Context,
+	lookback time.Duration,
+	entries []ProviderEntry,
+	services *Services,
+) error {
 	reliabilities, err := matureProviderReliabilities(entries)
 	if err != nil {
 		return err
@@ -312,6 +318,7 @@ func provisionPrewarm(ctx context.Context, lookback time.Duration, entries []Pro
 	if err != nil {
 		return err
 	}
+	services.SetPrewarmed(performances)
 
 	// The prewarm is DB-heavy (a fleet-wide reliability rebuild plus a score
 	// upsert). With a very large fleet connected, postgres can be transiently
@@ -425,8 +432,9 @@ func prewarmOnce(
 
 	// build network_client_location_reliability from the currently connected,
 	// latency/speed-tested providers
-	writeMatureProviderPerformances(ctx, performances)
+	writeMatureProviderConnectionPerformances(ctx, performances)
 	model.UpdateClientLocationReliabilities(ctx, now.Add(-lookback), now)
+	writeMatureProviderPerformanceSnapshot(ctx, performances)
 	writeMatureReliabilityScores(ctx, now, lookback, reliabilities)
 	return nil
 }
@@ -435,9 +443,9 @@ func prewarmOnce(
 // building the location-reliability snapshot. Completed tests belong to a
 // specific platform transport, so relying on an earlier transport's tests made
 // a newly active connection look untested and excluded it from matchmaking.
-// Writing the connection tables, instead of only the derived snapshot, also
-// makes later pipeline refreshes preserve the prewarmed evidence.
-func writeMatureProviderPerformances(
+// The provider-scoped snapshot restoration below handles later transport
+// replacement and pipeline refreshes.
+func writeMatureProviderConnectionPerformances(
 	ctx context.Context,
 	performances []matureProviderPerformance,
 ) {
@@ -504,6 +512,50 @@ func writeMatureProviderPerformances(
 	})
 }
 
+// Restores provider-scoped mature-market performance after a location refresh.
+// A fixture provider can replace its platform transport before the new
+// connection finishes tests; the derived snapshot would otherwise discard the
+// provider's established evidence and collapse the matchmaking pool.
+func writeMatureProviderPerformanceSnapshot(
+	ctx context.Context,
+	performances []matureProviderPerformance,
+) {
+	clientIds := make([]server.Id, 0, len(performances))
+	minRelativeLatencyMillisValues := make([]int64, 0, len(performances))
+	maxBytesPerSecondValues := make([]int64, 0, len(performances))
+	for _, performance := range performances {
+		clientIds = append(clientIds, performance.clientId)
+		minRelativeLatencyMillisValues = append(minRelativeLatencyMillisValues, performance.minRelativeLatencyMillis)
+		maxBytesPerSecondValues = append(maxBytesPerSecondValues, performance.maxBytesPerSecond)
+	}
+
+	server.Db(ctx, func(conn server.PgConn) {
+		server.RaisePgResult(conn.Exec(
+			ctx,
+			`
+			WITH mature_performance AS (
+				SELECT client_id, min_relative_latency_ms, max_bytes_per_second
+				FROM unnest($1::uuid[], $2::bigint[], $3::bigint[])
+					AS mature(client_id, min_relative_latency_ms, max_bytes_per_second)
+			)
+			UPDATE network_client_location_reliability
+			SET
+				min_relative_latency_ms = mature.min_relative_latency_ms::integer,
+				max_bytes_per_second = mature.max_bytes_per_second,
+				has_latency_test = true,
+				has_speed_test = true
+			FROM mature_performance mature
+			WHERE
+				network_client_location_reliability.client_id = mature.client_id AND
+				network_client_location_reliability.connected = true
+			`,
+			clientIds,
+			minRelativeLatencyMillisValues,
+			maxBytesPerSecondValues,
+		))
+	})
+}
+
 // Writes the seeded mature reliability beside the real connected/location and
 // latency/speed evidence. An inner join keeps disconnected providers out.
 func writeMatureReliabilityScores(

connect/sim-latency/provision_test.go

@@ -245,7 +245,7 @@ func TestMatureProviderPerformancesRejectInvalidGroundTruth(t *testing.T) {
 	}
 }
 
-func TestWriteMatureProviderPerformancesRestoresMatchmakingPool(t *testing.T) {
+func TestPrewarmedPipelineRestoresPerformanceAfterTransportReplacement(t *testing.T) {
 	server.DefaultTestEnv().Run(t, func(t testing.TB) {
 		ctx := context.Background()
 		providerNetworkId := server.NewId()
@@ -354,13 +354,39 @@ func TestWriteMatureProviderPerformancesRestoresMatchmakingPool(t *testing.T) {
 		if err != nil {
 			t.Fatalf("matureProviderPerformances: %v", err)
 		}
-		writeMatureProviderPerformances(ctx, performances)
+		writeMatureProviderConnectionPerformances(ctx, performances)
 		model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-13*time.Hour), server.NowUtc())
+		writeMatureProviderPerformanceSnapshot(ctx, performances)
 		after := findProviders()
 		if len(after.Providers) != 1 || after.Providers[0].ClientId != providerClientId {
 			t.Fatalf("matchmaking providers = %+v, want %s", after.Providers, providerClientId)
 		}
+		model.DisconnectNetworkClient(ctx, connectionId)
+		replacementConnectionId, _, _, _, err := model.ConnectNetworkClient(
+			ctx,
+			providerClientId,
+			"127.0.0.2:20001",
+			model.CreateNetworkClientHandler(ctx),
+		)
+		if err != nil {
+			t.Fatalf("replacement ConnectNetworkClient: %v", err)
+		}
+		if err := model.SetConnectionLocation(
+			ctx,
+			replacementConnectionId,
+			locationId,
+			&model.ConnectionLocationScores{},
+		); err != nil {
+			t.Fatalf("replacement SetConnectionLocation: %v", err)
+		}
 		model.UpdateClientLocationReliabilities(ctx, server.NowUtc().Add(-13*time.Hour), server.NowUtc())
+		withoutRestore := findProviders()
+		if len(withoutRestore.Providers) != 0 {
+			t.Fatalf("untested replacement entered matchmaking: %+v", withoutRestore.Providers)
+		}
+		services := &Services{}
+		services.SetPrewarmed(performances)
+		services.RunPipelineOnce(ctx)
 		afterRefresh := findProviders()
 		if len(afterRefresh.Providers) != 1 || afterRefresh.Providers[0].ClientId != providerClientId {
 			t.Fatalf("providers after pipeline refresh = %+v, want %s", afterRefresh.Providers, providerClientId)

connect/sim-latency/run.go

@@ -439,8 +439,7 @@ func Run(options *RunOptions) (retErr error) {
 	// propagate the redis sample export.
 	if 0 < options.Prewarm {
 		logf("prewarming: establishing the connected fleet (~%s reliability window)", options.Prewarm)
-		services.SetPrewarmed(true)
-		if err := provisionPrewarm(ctx, options.Prewarm, config.Fleet); err != nil {
+		if err := provisionPrewarm(ctx, options.Prewarm, config.Fleet, services); err != nil {
 			return phaseError(ctx, "prewarm_failed", "prewarm", err)
 		}
 		logf("prewarm complete; running pipeline")

connect/sim-latency/services.go

@@ -41,16 +41,32 @@ type Services struct {
 	runErr      error
 
 	// when prewarmed, the pipeline does not recompute reliability scores (which
-	// would overwrite the prewarmed scores); it only refreshes the location
-	// reliabilities (so churn gates selection) and re-exports the redis samples.
+	// would overwrite the prewarmed scores); it refreshes connected/location
+	// state, restores provider-scoped fixture performance, and re-exports the
+	// redis samples (so churn still gates selection).
 	prewarmed atomic.Bool
+
+	prewarmedPerformancesLock sync.RWMutex
+	prewarmedPerformances     []matureProviderPerformance
 }
 
 const servicesDrainTimeout = 15 * time.Second
 
-// SetPrewarmed switches the pipeline to prewarmed mode (see the field doc).
-func (self *Services) SetPrewarmed(prewarmed bool) {
-	self.prewarmed.Store(prewarmed)
+// SetPrewarmed switches the pipeline to prewarmed mode and freezes the fixture
+// performance evidence that each location refresh must restore. The pipeline
+// can already be running when warm-up reaches this point, so publish the copied
+// evidence before exposing the mode change.
+func (self *Services) SetPrewarmed(performances []matureProviderPerformance) {
+	self.prewarmedPerformancesLock.Lock()
+	self.prewarmedPerformances = append([]matureProviderPerformance(nil), performances...)
+	self.prewarmedPerformancesLock.Unlock()
+	self.prewarmed.Store(true)
+}
+
+func (self *Services) prewarmedPerformanceSnapshot() []matureProviderPerformance {
+	self.prewarmedPerformancesLock.RLock()
+	defer self.prewarmedPerformancesLock.RUnlock()
+	return append([]matureProviderPerformance(nil), self.prewarmedPerformances...)
 }
 
 // ServicesConfig configures the in-process environment.
@@ -281,9 +297,14 @@ func (self *Services) RunPipelineOnce(ctx context.Context) {
 	now := server.NowUtc()
 	if self.prewarmed.Load() {
 		// keep the location reliabilities fresh (churn -> connected state), then
-		// re-export the redis samples from the prewarmed scores. Do not recompute
-		// reliability scores (that would wipe the prewarm).
-		server.HandleError(func() { model.UpdateClientLocationReliabilities(ctx, now.Add(-12*time.Hour), now) })
+		// restore the fixture's initial performance evidence. Tests are attached
+		// to one short-lived platform transport, while the mature-market score is
+		// provider-scoped and must survive mechanical transport replacement. Do
+		// not recompute reliability scores (that would wipe the prewarm).
+		server.HandleError(func() {
+			model.UpdateClientLocationReliabilities(ctx, now.Add(-12*time.Hour), now)
+			writeMatureProviderPerformanceSnapshot(ctx, self.prewarmedPerformanceSnapshot())
+		})
 	} else {
 		server.HandleError(func() { model.RollupClientReliabilityStats(ctx, now) })
 		server.HandleError(func() { model.UpdateClientReliabilityScores(ctx, now, true) })