From 6508e1809fa358154030f93d99eb00cb6da3f652 Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:07:55 -0700 Subject: [PATCH 1/5] test: cover concurrent destination drains --- .../golden_acceptance_regression_test.go | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go index 5cd62fcdc..6888ee9dd 100644 --- a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go +++ b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go @@ -66,6 +66,33 @@ func TestGoldenMigrationSummaryPreservesRoutePropagationWindow(t *testing.T) { } } +func TestGoldenMigrationAllowsConcurrentDestinationConnectionDrain(t *testing.T) { + evidence := validGoldenMigrationTransitionEvidence( + "final-cutover", "front-migration-rollback", strings.Repeat("1", 64), strings.Repeat("2", 64), + ) + evidence.Destination.Before.PublicConnections = 14 + evidence.Destination.Before.GenerationConnections = 14 + evidence.Destination.After.PublicConnections = 10 + evidence.Destination.After.GenerationConnections = 10 + evidence.Destination.ConnectionCountDelta = -4 + + if err := validateGoldenMigrationTransition(evidence, "front-migration-cutover"); err != nil { + t.Fatalf("journal-correlated destination proof was rejected after concurrent drains: %v", err) + } + path := filepath.Join(t.TempDir(), "transition.json") + data, err := json.Marshal(evidence) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + validator := filepath.Join("..", "..", "deploy", "gcp", "validate-deploy-evidence.py") + if output, err := exec.Command("python3", validator, "--expect", "front-migration-cutover", path).CombinedOutput(); err != nil { + t.Fatalf("Python validator rejected journal-correlated destination proof after concurrent drains: %v: %s", err, output) + } +} + func TestGoldenMigrationTransitionRejectsRetargetedRoutingSelectors(t *testing.T) { evidence := validGoldenMigrationTransitionEvidence( "final-cutover", "front-migration-rollback", strings.Repeat("1", 64), strings.Repeat("2", 64), From 1e19407204e348f462a06631ec017bcde628a457 Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:11:23 -0700 Subject: [PATCH 2/5] fix: correlate migration proof by session --- .../golden_acceptance_regression_test.go | 17 ++++++++++++++++- .../golden_migration_evidence.go | 6 +++--- .../testdata/golden_fake_client/main.go | 2 +- deploy/gcp/switch-front-migration.sh | 12 ++++++++---- deploy/gcp/validate-deploy-evidence.py | 12 +++++++++--- 5 files changed, 37 insertions(+), 12 deletions(-) diff --git a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go index 6888ee9dd..cbaa09d94 100644 --- a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go +++ b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go @@ -91,6 +91,21 @@ func TestGoldenMigrationAllowsConcurrentDestinationConnectionDrain(t *testing.T) if output, err := exec.Command("python3", validator, "--expect", "front-migration-cutover", path).CombinedOutput(); err != nil { t.Fatalf("Python validator rejected journal-correlated destination proof after concurrent drains: %v: %s", err, output) } + + evidence.DestinationProof.JournalCorrelated = false + if err := validateGoldenMigrationTransition(evidence, "front-migration-cutover"); err == nil { + t.Fatal("Go validator accepted a destination proof without journal correlation") + } + data, err = json.Marshal(evidence) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + if output, err := exec.Command("python3", validator, "--expect", "front-migration-cutover", path).CombinedOutput(); err == nil { + t.Fatalf("Python validator accepted a destination proof without journal correlation: %s", output) + } } func TestGoldenMigrationTransitionRejectsRetargetedRoutingSelectors(t *testing.T) { @@ -958,7 +973,7 @@ func validGoldenMigrationTransitionEvidence(mode, priorType, priorSHA, preparati evidence.DestinationProof = goldenMigrationDestinationProof{ SHA256: strings.Repeat("8", 64), Challenge: strings.Repeat("9", 32), ConnectionID: strings.Repeat("a", 64), SessionID: "golden-session", - OriginalContinuityVerified: true, FreshPublicConnection: true, + OriginalContinuityVerified: true, FreshPublicConnection: true, JournalCorrelated: true, ObservedAt: "2026-08-02T00:00:01Z", ReceivedAt: "2026-08-02T00:00:01.5Z", } legacyMetrics := goldenMigrationLegacyMetrics{ diff --git a/cmd/subrouter-transport-observer/golden_migration_evidence.go b/cmd/subrouter-transport-observer/golden_migration_evidence.go index 7d606625f..4d70f567f 100644 --- a/cmd/subrouter-transport-observer/golden_migration_evidence.go +++ b/cmd/subrouter-transport-observer/golden_migration_evidence.go @@ -145,6 +145,7 @@ type goldenMigrationDestinationProof struct { SessionID string `json:"session_id"` OriginalContinuityVerified bool `json:"original_continuity_verified"` FreshPublicConnection bool `json:"fresh_public_connection"` + JournalCorrelated bool `json:"journal_correlated"` ObservedAt string `json:"observed_at"` ReceivedAt string `json:"received_at"` } @@ -466,9 +467,8 @@ func validateGoldenMigrationTransition(evidence *goldenMigrationEvidence, expect if evidence.Destination.Before.Generation == "" || evidence.Destination.Before.Generation != evidence.Destination.After.Generation || evidence.Destination.Before.InactiveConnections != 0 || evidence.Destination.After.InactiveConnections != 0 || - evidence.Destination.ConnectionCountDelta < 1 || evidence.Destination.After.GenerationConnections-evidence.Destination.Before.GenerationConnections != evidence.Destination.ConnectionCountDelta || - evidence.Destination.After.PublicConnections < evidence.Destination.Before.PublicConnections+1 { + evidence.Destination.After.PublicConnections < 1 || evidence.Destination.After.GenerationConnections < 1 { return failGolden("migration_destination_connection_count_invalid") } requested, activated, err := goldenPhaseDurationWithin( @@ -485,7 +485,7 @@ func validateGoldenMigrationTransition(evidence *goldenMigrationEvidence, expect !validGoldenSHA256(evidence.DestinationProof.SHA256) || !validGoldenChallenge(evidence.DestinationProof.Challenge) || !validGoldenSHA256(evidence.DestinationProof.ConnectionID) || !validGoldenOpaqueID(evidence.DestinationProof.SessionID) || !evidence.DestinationProof.OriginalContinuityVerified || - !evidence.DestinationProof.FreshPublicConnection { + !evidence.DestinationProof.FreshPublicConnection || !evidence.DestinationProof.JournalCorrelated { return failGolden("migration_destination_proof_invalid") } if err := validateGoldenMigrationMetrics(evidence); err != nil { diff --git a/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go b/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go index 7721654c4..fa127a187 100644 --- a/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go +++ b/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go @@ -235,7 +235,7 @@ func fakeAction(args []string) { "destination_proof": map[string]any{ "sha256": fmt.Sprintf("%x", proofDigest[:]), "challenge": challenge, "connection_id": proof.ConnectionID, "session_id": proof.SessionID, - "original_continuity_verified": true, "fresh_public_connection": true, + "original_continuity_verified": true, "fresh_public_connection": true, "journal_correlated": true, "observed_at": activated.Format(time.RFC3339Nano), "received_at": proofReceived.Format(time.RFC3339Nano), }, "source": map[string]any{ diff --git a/deploy/gcp/switch-front-migration.sh b/deploy/gcp/switch-front-migration.sh index 7462d97c6..82ea134e6 100755 --- a/deploy/gcp/switch-front-migration.sh +++ b/deploy/gcp/switch-front-migration.sh @@ -431,9 +431,13 @@ done [[ "${destination_correlated}" == true ]] \ || die "golden fresh public session did not reach the destination within ${proof_max_attempts} attempts" destination_after="$(supervisor_snapshot "${destination_kind}")" -jq -e --arg generation "${destination_generation}" --argjson before "${destination_before}" \ - '.generation == $generation and .public_connections >= ($before.public_connections + 1) and - .generation_connections >= ($before.generation_connections + 1) and .inactive_connections == 0' \ +# Earlier proof attempts drain asynchronously, so their departures can hide the +# accepted proof in an aggregate before/after delta. The journal match above +# correlates the exact session; this snapshot proves it is still live on the +# expected generation. +jq -e --arg generation "${destination_generation}" \ + '.generation == $generation and .public_connections >= 1 and + .generation_connections >= 1 and .inactive_connections == 0' \ < <(stream_shell_value "${destination_after}") >/dev/null \ || die "golden fresh public connection was not correlated to the destination generation" destination_connection_delta="$(( $(jq -r '.generation_connections' < <(stream_shell_value "${destination_after}")) - $(jq -r '.generation_connections' < <(stream_shell_value "${destination_before}")) ))" @@ -501,7 +505,7 @@ jq -n --arg schema 'subrouter.gcp.deploy-evidence/v1' --arg evidence_type "${evi evidence_emitted_at:$emitted_at}, destination_proof:{sha256:$proof_sha,challenge:$challenge,connection_id:$connection_id, session_id:$session_id, - original_continuity_verified:true,fresh_public_connection:true, + original_continuity_verified:true,fresh_public_connection:true,journal_correlated:true, observed_at:$activated_at,received_at:$proof_received_at}, destination:{before:$destination_before,after:$destination_after, connection_count_delta:$destination_delta}, diff --git a/deploy/gcp/validate-deploy-evidence.py b/deploy/gcp/validate-deploy-evidence.py index 86fad12d2..04696fc0a 100755 --- a/deploy/gcp/validate-deploy-evidence.py +++ b/deploy/gcp/validate-deploy-evidence.py @@ -933,6 +933,12 @@ def validate_front_migration_transition(document: dict[str, Any], expected: str) True, "destination_proof.original_continuity_verified", ) + exact( + boolean(field(proof, "journal_correlated", "destination_proof"), + "destination_proof.journal_correlated"), + True, + "destination_proof.journal_correlated", + ) exact( timestamp(field(proof, "observed_at", "destination_proof"), "destination_proof.observed_at"), activated, @@ -989,15 +995,15 @@ def validate_front_migration_transition(document: dict[str, Any], expected: str) delta = integer( field(destination, "connection_count_delta", "destination"), "destination.connection_count_delta", - minimum=1, + minimum=-(1 << 63), ) exact( delta, destination_after["generation_connections"] - destination_before["generation_connections"], "destination.connection_count_delta", ) - if destination_after["public_connections"] < destination_before["public_connections"] + 1: - fail("destination public connection count did not increase for the golden proof") + if destination_after["public_connections"] < 1 or destination_after["generation_connections"] < 1: + fail("destination had no live connection after the journal-correlated golden proof") metrics = obj(field(document, "metrics", "root"), "metrics") exact( field(metrics, "source_service", "metrics"), From e0d3b38089be276e4af75c868e5014ece3171824 Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:21:43 -0700 Subject: [PATCH 3/5] test: require post-snapshot migration liveness --- .../golden_acceptance_regression_test.go | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go index cbaa09d94..1c2f93996 100644 --- a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go +++ b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go @@ -108,6 +108,45 @@ func TestGoldenMigrationAllowsConcurrentDestinationConnectionDrain(t *testing.T) } } +func TestGoldenMigrationRejectsMissingPostSnapshotConnectionLiveness(t *testing.T) { + evidence := validGoldenMigrationTransitionEvidence( + "final-cutover", "front-migration-rollback", strings.Repeat("1", 64), strings.Repeat("2", 64), + ) + data, err := json.Marshal(evidence) + if err != nil { + t.Fatal(err) + } + var document map[string]any + if err := json.Unmarshal(data, &document); err != nil { + t.Fatal(err) + } + proof, ok := document["destination_proof"].(map[string]any) + if !ok { + t.Fatal("destination proof is not an object") + } + delete(proof, "post_snapshot_liveness") + data, err = json.Marshal(document) + if err != nil { + t.Fatal(err) + } + var missing goldenMigrationEvidence + if err := json.Unmarshal(data, &missing); err != nil { + t.Fatal(err) + } + if err := validateGoldenMigrationTransition(&missing, "front-migration-cutover"); err == nil { + t.Fatal("Go validator accepted migration evidence without exact post-snapshot connection liveness") + } + + path := filepath.Join(t.TempDir(), "transition.json") + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + validator := filepath.Join("..", "..", "deploy", "gcp", "validate-deploy-evidence.py") + if output, err := exec.Command("python3", validator, "--expect", "front-migration-cutover", path).CombinedOutput(); err == nil { + t.Fatalf("Python validator accepted migration evidence without exact post-snapshot connection liveness: %s", output) + } +} + func TestGoldenMigrationTransitionRejectsRetargetedRoutingSelectors(t *testing.T) { evidence := validGoldenMigrationTransitionEvidence( "final-cutover", "front-migration-rollback", strings.Repeat("1", 64), strings.Repeat("2", 64), From 5a62893fa97064687c2ff7e44c15b08afb4bbaff Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:33:25 -0700 Subject: [PATCH 4/5] fix: prove destination session stays live --- cmd/subrouter-transport-observer/golden.go | 1 + .../golden_acceptance_regression_test.go | 60 ++++++- .../golden_migration_evidence.go | 59 +++++- .../golden_migration_transition.go | 168 ++++++++++++++++-- .../testdata/golden_fake_client/main.go | 57 +++++- cmd/subrouter/deploy_scripts_test.go | 23 +++ deploy/gcp/deployment-contract.py | 49 +++++ deploy/gcp/switch-front-migration.sh | 53 +++++- deploy/gcp/validate-deploy-evidence.py | 58 +++++- 9 files changed, 497 insertions(+), 31 deletions(-) diff --git a/cmd/subrouter-transport-observer/golden.go b/cmd/subrouter-transport-observer/golden.go index 0a218461e..c3a53cf45 100644 --- a/cmd/subrouter-transport-observer/golden.go +++ b/cmd/subrouter-transport-observer/golden.go @@ -35,6 +35,7 @@ const ( goldenActionEvidenceLimit = 256 << 10 goldenActivationLimit = 30 * time.Second goldenMigrationPropagationLimit = 5 * time.Minute + goldenDestinationLivenessLimit = 10 * time.Second goldenBackendHealthStabilityLimit = 5 * time.Minute goldenRetirementLimit = 30 * time.Second goldenChunkGapFloor = 5 * time.Second diff --git a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go index 1c2f93996..9addf0bfe 100644 --- a/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go +++ b/cmd/subrouter-transport-observer/golden_acceptance_regression_test.go @@ -30,7 +30,10 @@ func TestGoldenMigrationUsesBoundedRoutePropagationWindow(t *testing.T) { evidence.Timestamps.ActivatedAt = "2026-08-02T00:04:59.000Z" evidence.DestinationProof.ObservedAt = evidence.Timestamps.ActivatedAt evidence.DestinationProof.ReceivedAt = "2026-08-02T00:04:59.500Z" - evidence.Timestamps.EvidenceEmittedAt = "2026-08-02T00:04:59.750Z" + evidence.DestinationProof.PostSnapshotLiveness.RequestedAt = "2026-08-02T00:04:59.600Z" + evidence.DestinationProof.PostSnapshotLiveness.ResponseChunkAt = "2026-08-02T00:04:59.700Z" + evidence.DestinationProof.PostSnapshotLiveness.ReceivedAt = "2026-08-02T00:04:59.800Z" + evidence.Timestamps.EvidenceEmittedAt = "2026-08-02T00:04:59.900Z" if err := validateGoldenMigrationTransition(evidence, "front-migration-cutover"); err != nil { t.Fatalf("sub-five-minute route propagation was rejected: %v", err) } @@ -38,7 +41,10 @@ func TestGoldenMigrationUsesBoundedRoutePropagationWindow(t *testing.T) { evidence.Timestamps.ActivatedAt = "2026-08-02T00:05:00.000Z" evidence.DestinationProof.ObservedAt = evidence.Timestamps.ActivatedAt evidence.DestinationProof.ReceivedAt = evidence.Timestamps.ActivatedAt - evidence.Timestamps.EvidenceEmittedAt = "2026-08-02T00:05:00.001Z" + evidence.DestinationProof.PostSnapshotLiveness.RequestedAt = "2026-08-02T00:05:00.001Z" + evidence.DestinationProof.PostSnapshotLiveness.ResponseChunkAt = "2026-08-02T00:05:00.002Z" + evidence.DestinationProof.PostSnapshotLiveness.ReceivedAt = "2026-08-02T00:05:00.003Z" + evidence.Timestamps.EvidenceEmittedAt = "2026-08-02T00:05:00.004Z" if err := validateGoldenMigrationTransition(evidence, "front-migration-cutover"); err == nil { t.Fatal("five-minute route propagation boundary was accepted") } @@ -51,7 +57,10 @@ func TestGoldenMigrationSummaryPreservesRoutePropagationWindow(t *testing.T) { evidence.Timestamps.ActivatedAt = "2026-08-02T00:04:59.123Z" evidence.DestinationProof.ObservedAt = evidence.Timestamps.ActivatedAt evidence.DestinationProof.ReceivedAt = "2026-08-02T00:04:59.500Z" - evidence.Timestamps.EvidenceEmittedAt = "2026-08-02T00:04:59.750Z" + evidence.DestinationProof.PostSnapshotLiveness.RequestedAt = "2026-08-02T00:04:59.600Z" + evidence.DestinationProof.PostSnapshotLiveness.ResponseChunkAt = "2026-08-02T00:04:59.700Z" + evidence.DestinationProof.PostSnapshotLiveness.ReceivedAt = "2026-08-02T00:04:59.800Z" + evidence.Timestamps.EvidenceEmittedAt = "2026-08-02T00:04:59.900Z" action := goldenActionSummary{} populateGoldenMigrationActionSummary(&action, evidence) @@ -75,6 +84,7 @@ func TestGoldenMigrationAllowsConcurrentDestinationConnectionDrain(t *testing.T) evidence.Destination.After.PublicConnections = 10 evidence.Destination.After.GenerationConnections = 10 evidence.Destination.ConnectionCountDelta = -4 + evidence.DestinationProof.PostSnapshotLiveness.DestinationSnapshotSHA256 = goldenMigrationSnapshotSHA256(evidence.Destination.After) if err := validateGoldenMigrationTransition(evidence, "front-migration-cutover"); err != nil { t.Fatalf("journal-correlated destination proof was rejected after concurrent drains: %v", err) @@ -147,6 +157,43 @@ func TestGoldenMigrationRejectsMissingPostSnapshotConnectionLiveness(t *testing. } } +func TestGoldenMigrationPostSnapshotLivenessUsesExactTransportConnection(t *testing.T) { + connectionID := strings.Repeat("a", 64) + otherConnectionID := strings.Repeat("b", 64) + requestID := "request-1" + boundary := time.Now().UTC() + stats := newObserverStats() + session := &goldenSession{ + observer: &runningGoldenObserver{stats: stats}, + done: make(chan struct{}), + } + stats.observe(transportEvent{ + Kind: "request_started", Path: "/v1/responses", RequestID: requestID, ConnectionID: connectionID, + }) + stats.observe(transportEvent{ + Kind: "response_chunk", Timestamp: boundary.Add(time.Millisecond).Format(time.RFC3339Nano), + Path: "/v1/responses", RequestID: requestID, ConnectionID: otherConnectionID, Bytes: 1, + }) + timedContext, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := waitGoldenConnectionResponseChunkAfter(timedContext, session, connectionID, boundary); err == nil { + t.Fatal("post-snapshot liveness accepted a response chunk from a different transport connection") + } + + want := boundary.Add(2 * time.Millisecond) + stats.observe(transportEvent{ + Kind: "response_chunk", Timestamp: want.Format(time.RFC3339Nano), + Path: "/v1/responses", RequestID: requestID, ConnectionID: connectionID, Bytes: 1, + }) + got, err := waitGoldenConnectionResponseChunkAfter(context.Background(), session, connectionID, boundary) + if err != nil { + t.Fatalf("exact transport response chunk was rejected: %v", err) + } + if !got.Equal(want) { + t.Fatalf("response chunk timestamp = %s, want %s", got, want) + } +} + func TestGoldenMigrationTransitionRejectsRetargetedRoutingSelectors(t *testing.T) { evidence := validGoldenMigrationTransitionEvidence( "final-cutover", "front-migration-rollback", strings.Repeat("1", 64), strings.Repeat("2", 64), @@ -1014,6 +1061,13 @@ func validGoldenMigrationTransitionEvidence(mode, priorType, priorSHA, preparati SessionID: "golden-session", OriginalContinuityVerified: true, FreshPublicConnection: true, JournalCorrelated: true, ObservedAt: "2026-08-02T00:00:01Z", ReceivedAt: "2026-08-02T00:00:01.5Z", + PostSnapshotLiveness: goldenMigrationPostSnapshotLiveness{ + SHA256: strings.Repeat("b", 64), Challenge: strings.Repeat("c", 32), + ConnectionID: strings.Repeat("a", 64), SessionID: "golden-session", + DestinationSnapshotSHA256: goldenMigrationSnapshotSHA256(evidence.Destination.After), + RequestedAt: "2026-08-02T00:00:01.6Z", ResponseChunkAt: "2026-08-02T00:00:01.7Z", + ReceivedAt: "2026-08-02T00:00:01.8Z", + }, } legacyMetrics := goldenMigrationLegacyMetrics{ NRestarts: goldenDeployCounter{Before: goldenInt64(0), After: goldenInt64(0)}, diff --git a/cmd/subrouter-transport-observer/golden_migration_evidence.go b/cmd/subrouter-transport-observer/golden_migration_evidence.go index 4d70f567f..68d7d7f56 100644 --- a/cmd/subrouter-transport-observer/golden_migration_evidence.go +++ b/cmd/subrouter-transport-observer/golden_migration_evidence.go @@ -139,15 +139,27 @@ type goldenMigrationTimestamps struct { } type goldenMigrationDestinationProof struct { - SHA256 string `json:"sha256"` - Challenge string `json:"challenge"` - ConnectionID string `json:"connection_id"` - SessionID string `json:"session_id"` - OriginalContinuityVerified bool `json:"original_continuity_verified"` - FreshPublicConnection bool `json:"fresh_public_connection"` - JournalCorrelated bool `json:"journal_correlated"` - ObservedAt string `json:"observed_at"` - ReceivedAt string `json:"received_at"` + SHA256 string `json:"sha256"` + Challenge string `json:"challenge"` + ConnectionID string `json:"connection_id"` + SessionID string `json:"session_id"` + OriginalContinuityVerified bool `json:"original_continuity_verified"` + FreshPublicConnection bool `json:"fresh_public_connection"` + JournalCorrelated bool `json:"journal_correlated"` + ObservedAt string `json:"observed_at"` + ReceivedAt string `json:"received_at"` + PostSnapshotLiveness goldenMigrationPostSnapshotLiveness `json:"post_snapshot_liveness"` +} + +type goldenMigrationPostSnapshotLiveness struct { + SHA256 string `json:"sha256"` + Challenge string `json:"challenge"` + ConnectionID string `json:"connection_id"` + SessionID string `json:"session_id"` + DestinationSnapshotSHA256 string `json:"destination_snapshot_sha256"` + RequestedAt string `json:"requested_at"` + ResponseChunkAt string `json:"response_chunk_at"` + ReceivedAt string `json:"received_at"` } type goldenMigrationSnapshot struct { @@ -488,6 +500,20 @@ func validateGoldenMigrationTransition(evidence *goldenMigrationEvidence, expect !evidence.DestinationProof.FreshPublicConnection || !evidence.DestinationProof.JournalCorrelated { return failGolden("migration_destination_proof_invalid") } + liveness := evidence.DestinationProof.PostSnapshotLiveness + livenessRequested, requestErr := parseGoldenEvidenceTime(liveness.RequestedAt) + livenessChunk, chunkErr := parseGoldenEvidenceTime(liveness.ResponseChunkAt) + livenessReceived, receivedErr := parseGoldenEvidenceTime(liveness.ReceivedAt) + if requestErr != nil || chunkErr != nil || receivedErr != nil || + !validGoldenSHA256(liveness.SHA256) || !validGoldenChallenge(liveness.Challenge) || + liveness.ConnectionID != evidence.DestinationProof.ConnectionID || + liveness.SessionID != evidence.DestinationProof.SessionID || + liveness.DestinationSnapshotSHA256 != goldenMigrationSnapshotSHA256(evidence.Destination.After) || + livenessRequested.Before(proofReceived) || livenessChunk.Before(livenessRequested) || + livenessReceived.Before(livenessChunk) || emitted.Before(livenessReceived) || + livenessReceived.Sub(livenessRequested) >= goldenDestinationLivenessLimit { + return failGolden("migration_destination_liveness_invalid") + } if err := validateGoldenMigrationMetrics(evidence); err != nil { return err } @@ -499,6 +525,21 @@ func validateGoldenMigrationTransition(evidence *goldenMigrationEvidence, expect return nil } +func goldenMigrationSnapshotSHA256(snapshot goldenMigrationSnapshot) string { + canonical, err := json.Marshal(map[string]any{ + "generation": snapshot.Generation, + "generation_connections": snapshot.GenerationConnections, + "inactive_connections": snapshot.InactiveConnections, + "kind": snapshot.Kind, + "public_connections": snapshot.PublicConnections, + }) + if err != nil { + return "" + } + digest := sha256.Sum256(canonical) + return hex.EncodeToString(digest[:]) +} + func validateGoldenMigrationMetrics(evidence *goldenMigrationEvidence) error { wantSource, wantDestination := "slot", "legacy" if evidence.Routing.Before == "legacy" { diff --git a/cmd/subrouter-transport-observer/golden_migration_transition.go b/cmd/subrouter-transport-observer/golden_migration_transition.go index 1a856673c..e75bcede0 100644 --- a/cmd/subrouter-transport-observer/golden_migration_transition.go +++ b/cmd/subrouter-transport-observer/golden_migration_transition.go @@ -13,9 +13,11 @@ import ( ) const ( - goldenDestinationProofRequestSchema = "subrouter.gcp.destination-proof-request/v1" - goldenDestinationProofSchema = "subrouter.gcp.destination-proof/v1" - goldenDestinationProofMaxAttempts = 64 + goldenDestinationProofRequestSchema = "subrouter.gcp.destination-proof-request/v1" + goldenDestinationProofSchema = "subrouter.gcp.destination-proof/v1" + goldenDestinationLivenessRequestSchema = "subrouter.gcp.destination-liveness-request/v1" + goldenDestinationLivenessProofSchema = "subrouter.gcp.destination-liveness/v1" + goldenDestinationProofMaxAttempts = 64 ) type goldenDestinationProofRequest struct { @@ -48,6 +50,31 @@ type goldenDestinationProof struct { ObservedAt string `json:"observed_at"` } +type goldenDestinationLivenessRequest struct { + Schema string `json:"schema"` + Challenge string `json:"challenge"` + Operation string `json:"operation"` + Destination string `json:"destination"` + DestinationGeneration string `json:"destination_generation"` + ConnectionID string `json:"connection_id"` + SessionID string `json:"session_id"` + DestinationSnapshotSHA256 string `json:"destination_snapshot_sha256"` + RequestedAt string `json:"requested_at"` +} + +type goldenDestinationLivenessProof struct { + Schema string `json:"schema"` + Challenge string `json:"challenge"` + Operation string `json:"operation"` + Destination string `json:"destination"` + DestinationGeneration string `json:"destination_generation"` + ConnectionID string `json:"connection_id"` + SessionID string `json:"session_id"` + DestinationSnapshotSHA256 string `json:"destination_snapshot_sha256"` + RequestedAt string `json:"requested_at"` + ResponseChunkAt string `json:"response_chunk_at"` +} + func (r *goldenRunner) runMigrationTransitionWithProof( ctx context.Context, operation string, @@ -97,10 +124,15 @@ func (r *goldenRunner) runMigrationTransitionWithProof( var request goldenDestinationProofRequest var proof goldenDestinationProof var proofDigest [sha256.Size]byte + var livenessRequest goldenDestinationLivenessRequest + var livenessProof goldenDestinationLivenessProof + var livenessProofDigest [sha256.Size]byte var destinationSession *goldenSession for attempt := 1; attempt <= goldenDestinationProofMaxAttempts; attempt++ { attemptRequestPath := goldenMigrationAttemptPath(requestPath, attempt) attemptProofPath := goldenMigrationAttemptPath(proofPath, attempt) + attemptLivenessRequestPath := attemptProofPath + ".liveness-request" + attemptLivenessProofPath := attemptProofPath + ".liveness-proof" requestData, err := waitGoldenHandshakeFile(ctx, attemptRequestPath, command.done) if err != nil { return goldenActionSummary{}, nil, err @@ -166,8 +198,8 @@ func (r *goldenRunner) runMigrationTransitionWithProof( if err != nil || time.Since(requested) >= goldenMigrationPropagationLimit || writePrivateFile(attemptProofPath, proofFileData) != nil { return goldenActionSummary{}, nil, failGolden("migration_destination_proof_write_failed") } - retry, err := waitGoldenMigrationAttemptOutcome( - ctx, goldenMigrationAttemptPath(requestPath, attempt+1), command.done, + retry, livenessData, err := waitGoldenMigrationAttemptOutcome( + ctx, goldenMigrationAttemptPath(requestPath, attempt+1), attemptLivenessRequestPath, command.done, ) if err != nil { return goldenActionSummary{}, nil, err @@ -182,6 +214,42 @@ func (r *goldenRunner) runMigrationTransitionWithProof( } continue } + livenessRequest = goldenDestinationLivenessRequest{} + if err := json.Unmarshal(livenessData, &livenessRequest); err != nil || + livenessRequest.Schema != goldenDestinationLivenessRequestSchema || + !validGoldenChallenge(livenessRequest.Challenge) || livenessRequest.Operation != operation || + livenessRequest.Destination != destination || + livenessRequest.DestinationGeneration != request.DestinationGeneration || + livenessRequest.ConnectionID != proof.ConnectionID || livenessRequest.SessionID != proof.SessionID || + !validGoldenSHA256(livenessRequest.DestinationSnapshotSHA256) { + return goldenActionSummary{}, nil, failGolden("migration_destination_liveness_request_invalid") + } + livenessRequested, err := parseGoldenEvidenceTime(livenessRequest.RequestedAt) + proofObserved, proofTimeErr := parseGoldenEvidenceTime(proof.ObservedAt) + if err != nil || proofTimeErr != nil || livenessRequested.Before(proofObserved) || + time.Since(livenessRequested) >= goldenDestinationLivenessLimit { + return goldenActionSummary{}, nil, failGolden("migration_destination_liveness_request_invalid") + } + responseChunkAt, err := waitGoldenConnectionResponseChunkAfter( + ctx, candidateSession, proof.ConnectionID, livenessRequested, + ) + if err != nil { + return goldenActionSummary{}, nil, err + } + livenessProof = goldenDestinationLivenessProof{ + Schema: goldenDestinationLivenessProofSchema, Challenge: livenessRequest.Challenge, + Operation: operation, Destination: destination, + DestinationGeneration: request.DestinationGeneration, + ConnectionID: proof.ConnectionID, SessionID: proof.SessionID, + DestinationSnapshotSHA256: livenessRequest.DestinationSnapshotSHA256, + RequestedAt: livenessRequest.RequestedAt, ResponseChunkAt: responseChunkAt.Format(time.RFC3339Nano), + } + livenessProofData, err := json.Marshal(livenessProof) + livenessProofFileData := append(livenessProofData, '\n') + livenessProofDigest = sha256.Sum256(livenessProofFileData) + if err != nil || writePrivateFile(attemptLivenessProofPath, livenessProofFileData) != nil { + return goldenActionSummary{}, nil, failGolden("migration_destination_liveness_proof_write_failed") + } destinationSession = candidateSession break } @@ -207,7 +275,14 @@ func (r *goldenRunner) runMigrationTransitionWithProof( evidence.Routing.After != destination || evidence.Timestamps.TransitionRequestedAt != request.TransitionRequestedAt || evidence.Timestamps.ActivatedAt != proof.ObservedAt || evidence.DestinationProof.SHA256 != fmt.Sprintf("%x", proofDigest[:]) || evidence.DestinationProof.Challenge != proof.Challenge || evidence.DestinationProof.ConnectionID != proof.ConnectionID || - evidence.DestinationProof.SessionID != proof.SessionID { + evidence.DestinationProof.SessionID != proof.SessionID || + evidence.DestinationProof.PostSnapshotLiveness.SHA256 != fmt.Sprintf("%x", livenessProofDigest[:]) || + evidence.DestinationProof.PostSnapshotLiveness.Challenge != livenessProof.Challenge || + evidence.DestinationProof.PostSnapshotLiveness.ConnectionID != livenessProof.ConnectionID || + evidence.DestinationProof.PostSnapshotLiveness.SessionID != livenessProof.SessionID || + evidence.DestinationProof.PostSnapshotLiveness.DestinationSnapshotSHA256 != livenessProof.DestinationSnapshotSHA256 || + evidence.DestinationProof.PostSnapshotLiveness.RequestedAt != livenessProof.RequestedAt || + evidence.DestinationProof.PostSnapshotLiveness.ResponseChunkAt != livenessProof.ResponseChunkAt { return result, nil, failGolden("migration_destination_proof_evidence_mismatch") } if sessionDone(destinationSession) { @@ -232,25 +307,94 @@ func goldenSessionThreadID(session *goldenSession) string { return session.threadID } -func waitGoldenMigrationAttemptOutcome(ctx context.Context, nextRequestPath string, commandDone <-chan struct{}) (bool, error) { +func waitGoldenConnectionResponseChunkAfter( + ctx context.Context, + session *goldenSession, + connectionID string, + boundary time.Time, +) (time.Time, error) { + if session == nil || session.observer == nil || session.observer.stats == nil || + !validGoldenSHA256(connectionID) || boundary.IsZero() { + return time.Time{}, failGolden("migration_destination_connection_not_held") + } + waitContext, cancel := context.WithTimeout(ctx, goldenDestinationLivenessLimit) + defer cancel() + for { + requests, chunks, observerErrors := session.observer.stats.snapshot() + if observerErrors != 0 { + return time.Time{}, failGolden("migration_destination_connection_not_held") + } + requestIDs := make(map[string]bool) + for _, request := range requests { + if request.ConnectionID == connectionID && + (request.Path == "/v1/responses" || request.Path == "/responses") { + requestIDs[request.RequestID] = true + } + } + for _, chunk := range chunks { + if chunk.Kind != "response_chunk" || chunk.ConnectionID != connectionID || chunk.Bytes <= 0 || + !requestIDs[chunk.RequestID] || (chunk.Path != "/v1/responses" && chunk.Path != "/responses") { + continue + } + stamp, err := time.Parse(time.RFC3339Nano, chunk.Timestamp) + if err != nil || stamp.IsZero() { + return time.Time{}, failGolden("migration_destination_connection_not_held") + } + if !stamp.Before(boundary) { + return stamp, nil + } + } + if sessionDone(session) { + return time.Time{}, failGolden("migration_destination_connection_not_held") + } + select { + case <-waitContext.Done(): + return time.Time{}, failGolden("migration_destination_connection_not_held") + case <-session.done: + return time.Time{}, failGolden("migration_destination_connection_not_held") + case <-session.observer.stats.notify: + } + } +} + +func waitGoldenMigrationAttemptOutcome( + ctx context.Context, + nextRequestPath string, + livenessRequestPath string, + commandDone <-chan struct{}, +) (bool, []byte, error) { ticker := time.NewTicker(10 * time.Millisecond) defer ticker.Stop() for { info, err := os.Lstat(nextRequestPath) if err == nil { if !info.Mode().IsRegular() || info.Size() <= 0 || info.Size() > goldenActionEvidenceLimit { - return false, failGolden("deployment_handshake_file_invalid") + return false, nil, failGolden("deployment_handshake_file_invalid") + } + return true, nil, nil + } + if !errors.Is(err, os.ErrNotExist) { + return false, nil, failGolden("deployment_handshake_file_invalid") + } + info, err = os.Lstat(livenessRequestPath) + if err == nil { + if !info.Mode().IsRegular() || info.Size() <= 0 || info.Size() > goldenActionEvidenceLimit { + return false, nil, failGolden("deployment_handshake_file_invalid") + } + data, readErr := os.ReadFile(livenessRequestPath) + if readErr != nil { + return false, nil, failGolden("deployment_handshake_file_invalid") } - return true, nil + return false, data, nil } if !errors.Is(err, os.ErrNotExist) { - return false, failGolden("deployment_handshake_file_invalid") + return false, nil, failGolden("deployment_handshake_file_invalid") } select { case <-ctx.Done(): - return false, ctx.Err() + return false, nil, ctx.Err() case <-commandDone: - return false, nil + return false, nil, failGolden("migration_destination_liveness_request_missing") case <-ticker.C: } } diff --git a/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go b/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go index fa127a187..4626a265d 100644 --- a/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go +++ b/cmd/subrouter-transport-observer/testdata/golden_fake_client/main.go @@ -203,6 +203,53 @@ func fakeAction(args []string) { } proofReceived := time.Now().UTC() proofDigest := sha256.Sum256(proofData) + destinationAfter := map[string]any{ + "kind": destination, "generation": destinationGeneration, + "public_connections": 1, "generation_connections": 1, "inactive_connections": 0, + } + destinationSnapshotData, err := json.Marshal(destinationAfter) + if err != nil { + os.Exit(9) + } + destinationSnapshotDigest := sha256.Sum256(destinationSnapshotData) + destinationSnapshotSHA := fmt.Sprintf("%x", destinationSnapshotDigest[:]) + livenessChallenge := strings.Repeat("7", 32) + livenessRequested := time.Now().UTC() + livenessRequest := map[string]any{ + "schema": "subrouter.gcp.destination-liveness-request/v1", "challenge": livenessChallenge, + "operation": migrationOperation, "destination": destination, + "destination_generation": destinationGeneration, "connection_id": proof.ConnectionID, + "session_id": proof.SessionID, "destination_snapshot_sha256": destinationSnapshotSHA, + "requested_at": livenessRequested.Format(time.RFC3339Nano), + } + livenessRequestPath := proofPath + ".liveness-request" + livenessProofPath := proofPath + ".liveness-proof" + if fakeWriteJSON(livenessRequestPath, livenessRequest) != nil { + os.Exit(9) + } + livenessProofData, err := fakeWaitFile(livenessProofPath, 10*time.Second) + var livenessProof struct { + Schema string `json:"schema"` + Challenge string `json:"challenge"` + ConnectionID string `json:"connection_id"` + SessionID string `json:"session_id"` + DestinationSnapshotSHA256 string `json:"destination_snapshot_sha256"` + RequestedAt string `json:"requested_at"` + ResponseChunkAt string `json:"response_chunk_at"` + } + if err != nil || json.Unmarshal(livenessProofData, &livenessProof) != nil || + livenessProof.Schema != "subrouter.gcp.destination-liveness/v1" || + livenessProof.Challenge != livenessChallenge || livenessProof.ConnectionID != proof.ConnectionID || + livenessProof.SessionID != proof.SessionID || + livenessProof.DestinationSnapshotSHA256 != destinationSnapshotSHA || + livenessProof.RequestedAt != livenessRequested.Format(time.RFC3339Nano) { + os.Exit(9) + } + livenessReceived := time.Now().UTC() + if _, err := time.Parse(time.RFC3339Nano, livenessProof.ResponseChunkAt); err != nil { + os.Exit(9) + } + livenessProofDigest := sha256.Sum256(livenessProofData) predecessorLinux := "99fcd10d912184c160370eb228b382795101f2b5b2467244f995aa2d10b0c323" legacy := map[string]any{"service": "subrouter.service", "generation": "legacy-generation", "checksum": predecessorLinux} front := map[string]any{"slot": "slot-a", "generation": "front-generation", "checksum": candidate, "control_checksum": candidate, "worker_checksum": goldenFakeBootstrapSHA256} @@ -237,13 +284,21 @@ func fakeAction(args []string) { "connection_id": proof.ConnectionID, "session_id": proof.SessionID, "original_continuity_verified": true, "fresh_public_connection": true, "journal_correlated": true, "observed_at": activated.Format(time.RFC3339Nano), "received_at": proofReceived.Format(time.RFC3339Nano), + "post_snapshot_liveness": map[string]any{ + "sha256": fmt.Sprintf("%x", livenessProofDigest[:]), "challenge": livenessChallenge, + "connection_id": proof.ConnectionID, "session_id": proof.SessionID, + "destination_snapshot_sha256": destinationSnapshotSHA, + "requested_at": livenessRequested.Format(time.RFC3339Nano), + "response_chunk_at": livenessProof.ResponseChunkAt, + "received_at": livenessReceived.Format(time.RFC3339Nano), + }, }, "source": map[string]any{ "before": snapshot(source, sourceGeneration, expected), "after": snapshot(source, sourceGeneration, expected), "accepting_new_public_before": true, "accepting_new_public_after": false, }, "destination": map[string]any{ - "before": snapshot(destination, destinationGeneration, 0), "after": snapshot(destination, destinationGeneration, 1), + "before": snapshot(destination, destinationGeneration, 0), "after": destinationAfter, "connection_count_delta": 1, }, "metrics": map[string]any{ diff --git a/cmd/subrouter/deploy_scripts_test.go b/cmd/subrouter/deploy_scripts_test.go index d91a0ad67..e2b6a219e 100644 --- a/cmd/subrouter/deploy_scripts_test.go +++ b/cmd/subrouter/deploy_scripts_test.go @@ -1863,6 +1863,29 @@ func TestDeploymentContractValidatesGoldenTransitionProofs(t *testing.T) { if output, err := run(proofArgs...); err == nil { t.Fatalf("empty destination session succeeded: %s", output) } + + liveness := filepath.Join(t.TempDir(), "liveness.json") + connectionID := strings.Repeat("b", 64) + snapshotSHA := strings.Repeat("c", 64) + livenessBody := fmt.Sprintf(`{"schema":"subrouter.gcp.destination-liveness/v1","challenge":"challenge","operation":"final-cutover","destination":"front","destination_generation":"generation-b","connection_id":%q,"session_id":"session-id","destination_snapshot_sha256":%q,"requested_at":"2026-08-03T10:01:00Z","response_chunk_at":"2026-08-03T10:01:00.500Z"}`, connectionID, snapshotSHA) + if err := os.WriteFile(liveness, []byte(livenessBody), 0o600); err != nil { + t.Fatal(err) + } + livenessArgs := []string{ + "validate-destination-liveness", liveness, "challenge", "final-cutover", "front", "generation-b", + connectionID, "session-id", snapshotSHA, "2026-08-03T10:01:00Z", "2026-08-03T10:01:00.750Z", + } + if output, err := run(livenessArgs...); err != nil || len(output) != 0 { + t.Fatalf("destination liveness result = %q, %v", output, err) + } + lateLiveness := strings.Replace(livenessBody, "10:01:00.500Z", "10:01:10.000Z", 1) + if err := os.WriteFile(liveness, []byte(lateLiveness), 0o600); err != nil { + t.Fatal(err) + } + livenessArgs[len(livenessArgs)-1] = "2026-08-03T10:01:10.000Z" + if output, err := run(livenessArgs...); err == nil { + t.Fatalf("ten-second destination liveness boundary succeeded: %s", output) + } } func TestDeploymentContractProbesSlotEndpoint(t *testing.T) { diff --git a/deploy/gcp/deployment-contract.py b/deploy/gcp/deployment-contract.py index b4e8c9fba..4ed857f0a 100755 --- a/deploy/gcp/deployment-contract.py +++ b/deploy/gcp/deployment-contract.py @@ -436,6 +436,42 @@ def command_validate_destination_proof(args: argparse.Namespace) -> None: ) +def command_validate_destination_liveness(args: argparse.Namespace) -> None: + if re.fullmatch(r"[0-9a-f]{64}", args.connection_id) is None: + fail("destination liveness connection_id is invalid") + if re.fullmatch(r"[A-Za-z0-9._:-]{1,256}", args.session_id) is None: + fail("destination liveness session_id is invalid") + if re.fullmatch(r"[0-9a-f]{64}", args.destination_snapshot_sha256) is None: + fail("destination liveness snapshot SHA-256 is invalid") + document = load_json(args.path, "destination liveness proof") + expect_exact( + document, + { + "schema": "subrouter.gcp.destination-liveness/v1", + "challenge": args.challenge, + "operation": args.operation, + "destination": args.destination, + "destination_generation": args.destination_generation, + "connection_id": args.connection_id, + "session_id": args.session_id, + "destination_snapshot_sha256": args.destination_snapshot_sha256, + "requested_at": args.requested_at, + }, + "destination liveness proof", + ) + requested = parse_timestamp(args.requested_at, "destination liveness requested time") + response_chunk = parse_timestamp(document.get("response_chunk_at"), "destination liveness response chunk time") + received = parse_timestamp(args.received_at, "destination liveness received time") + validate_ordered_window( + requested, + requested, + response_chunk, + received, + "destination liveness proof", + timedelta(seconds=10), + ) + + def add_path(parser: argparse.ArgumentParser, name: str) -> None: parser.add_argument(name, type=Path) @@ -529,6 +565,19 @@ def build_parser() -> argparse.ArgumentParser: proof.add_argument("received_at") proof.set_defaults(handler=command_validate_destination_proof) + liveness = commands.add_parser("validate-destination-liveness") + add_path(liveness, "path") + liveness.add_argument("challenge") + liveness.add_argument("operation") + liveness.add_argument("destination") + liveness.add_argument("destination_generation") + liveness.add_argument("connection_id") + liveness.add_argument("session_id") + liveness.add_argument("destination_snapshot_sha256") + liveness.add_argument("requested_at") + liveness.add_argument("received_at") + liveness.set_defaults(handler=command_validate_destination_liveness) + return parser diff --git a/deploy/gcp/switch-front-migration.sh b/deploy/gcp/switch-front-migration.sh index 82ea134e6..f81cfe08e 100755 --- a/deploy/gcp/switch-front-migration.sh +++ b/deploy/gcp/switch-front-migration.sh @@ -43,6 +43,7 @@ ARTIFACT_DIR="${SUBROUTER_DEPLOY_ARTIFACT_DIR:-${PWD}/artifacts/gcp-front-transi RUN_LABEL="${GITHUB_RUN_ID:-local}-${GITHUB_RUN_ATTEMPT:-0}-lb-${OPERATION}-$$" RUN_LABEL="${RUN_LABEL//[^a-zA-Z0-9._-]/-}" MIGRATION_PROPAGATION_LIMIT_MS=300000 +DESTINATION_LIVENESS_LIMIT_MS=10000 REMOTE_INSTALLER="/tmp/install-front-slots-${RUN_LABEL}.sh" REMOTE_DEPLOYMENT_CONTRACT="/tmp/deployment-contract-${RUN_LABEL}.py" SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" @@ -381,7 +382,11 @@ destination_correlated=false while (( proof_attempt <= proof_max_attempts )); do proof_request_path="$(proof_attempt_path "${DESTINATION_PROOF_REQUEST}" "${proof_attempt}")" proof_path="$(proof_attempt_path "${DESTINATION_PROOF}" "${proof_attempt}")" - [[ ! -e "${proof_request_path}" && ! -L "${proof_request_path}" && ! -e "${proof_path}" && ! -L "${proof_path}" ]] \ + liveness_request_path="${proof_path}.liveness-request" + liveness_proof_path="${proof_path}.liveness-proof" + [[ ! -e "${proof_request_path}" && ! -L "${proof_request_path}" && ! -e "${proof_path}" && ! -L "${proof_path}" && + ! -e "${liveness_request_path}" && ! -L "${liveness_request_path}" && + ! -e "${liveness_proof_path}" && ! -L "${liveness_proof_path}" ]] \ || die "destination proof attempt path already exists" proof_challenge="$(python3 -c 'import secrets; print(secrets.token_hex(16))')" proof_request_tmp="$(mktemp "${proof_request_path}.tmp.XXXXXX")" @@ -432,14 +437,44 @@ done || die "golden fresh public session did not reach the destination within ${proof_max_attempts} attempts" destination_after="$(supervisor_snapshot "${destination_kind}")" # Earlier proof attempts drain asynchronously, so their departures can hide the -# accepted proof in an aggregate before/after delta. The journal match above -# correlates the exact session; this snapshot proves it is still live on the -# expected generation. +# accepted proof in an aggregate before/after delta. Require a live connection +# on the exact generation, then challenge the golden observer to produce a +# response chunk on the accepted transport after this snapshot. jq -e --arg generation "${destination_generation}" \ '.generation == $generation and .public_connections >= 1 and .generation_connections >= 1 and .inactive_connections == 0' \ < <(stream_shell_value "${destination_after}") >/dev/null \ || die "golden fresh public connection was not correlated to the destination generation" +destination_snapshot_canonical="$(printf '%s' "${destination_after}" | jq -cS .)" +destination_snapshot_sha256="$(printf '%s' "${destination_snapshot_canonical}" | sha256sum | awk '{print $1}')" +liveness_challenge="$(python3 -c 'import secrets; print(secrets.token_hex(16))')" +liveness_requested_at="$(utc_now)" +liveness_requested_ms="$(epoch_millis)" +liveness_request_tmp="$(mktemp "${liveness_request_path}.tmp.XXXXXX")" +jq -n --arg schema 'subrouter.gcp.destination-liveness-request/v1' \ + --arg challenge "${liveness_challenge}" --arg operation "${OPERATION}" \ + --arg destination "${destination_kind}" --arg generation "${destination_generation}" \ + --arg connection_id "${destination_connection_id}" --arg session_id "${destination_session_id}" \ + --arg snapshot_sha "${destination_snapshot_sha256}" --arg requested_at "${liveness_requested_at}" \ + '{schema:$schema,challenge:$challenge,operation:$operation,destination:$destination, + destination_generation:$generation,connection_id:$connection_id,session_id:$session_id, + destination_snapshot_sha256:$snapshot_sha,requested_at:$requested_at}' \ + >"${liveness_request_tmp}" +chmod 0600 "${liveness_request_tmp}" +mv -f -- "${liveness_request_tmp}" "${liveness_request_path}" +while [[ ! -f "${liveness_proof_path}" || -L "${liveness_proof_path}" ]]; do + (( $(epoch_millis) - liveness_requested_ms < DESTINATION_LIVENESS_LIMIT_MS )) \ + || die "golden destination connection produced no response chunk after the destination snapshot" + sleep 0.05 +done +liveness_received_at="$(utc_now)" +python3 "${DEPLOYMENT_CONTRACT}" validate-destination-liveness \ + "${liveness_proof_path}" "${liveness_challenge}" "${OPERATION}" \ + "${destination_kind}" "${destination_generation}" "${destination_connection_id}" \ + "${destination_session_id}" "${destination_snapshot_sha256}" "${liveness_requested_at}" \ + "${liveness_received_at}" +liveness_proof_sha256="$(sha256sum "${liveness_proof_path}" | awk '{print $1}')" +liveness_response_chunk_at="$(jq -r '.response_chunk_at' "${liveness_proof_path}")" destination_connection_delta="$(( $(jq -r '.generation_connections' < <(stream_shell_value "${destination_after}")) - $(jq -r '.generation_connections' < <(stream_shell_value "${destination_before}")) ))" legacy_restarts_after="$(service_restarts legacy)" legacy_oom_after="$(service_oom_kills legacy)" @@ -480,6 +515,10 @@ jq -n --arg schema 'subrouter.gcp.deploy-evidence/v1' --arg evidence_type "${evi --arg proof_received_at "${proof_received_at}" --arg emitted_at "${evidence_emitted_at}" \ --arg proof_sha "${destination_proof_sha256}" --arg challenge "${proof_challenge}" \ --arg connection_id "${destination_connection_id}" --arg session_id "${destination_session_id}" \ + --arg liveness_sha "${liveness_proof_sha256}" --arg liveness_challenge "${liveness_challenge}" \ + --arg snapshot_sha "${destination_snapshot_sha256}" --arg liveness_requested_at "${liveness_requested_at}" \ + --arg liveness_response_chunk_at "${liveness_response_chunk_at}" \ + --arg liveness_received_at "${liveness_received_at}" \ --argjson before "${source_before}" \ --argjson after "${source_after}" --argjson expected "${EXPECTED_CONNECTIONS}" \ --argjson destination_before "${destination_before}" --argjson destination_after "${destination_after}" \ @@ -506,7 +545,11 @@ jq -n --arg schema 'subrouter.gcp.deploy-evidence/v1' --arg evidence_type "${evi destination_proof:{sha256:$proof_sha,challenge:$challenge,connection_id:$connection_id, session_id:$session_id, original_continuity_verified:true,fresh_public_connection:true,journal_correlated:true, - observed_at:$activated_at,received_at:$proof_received_at}, + observed_at:$activated_at,received_at:$proof_received_at, + post_snapshot_liveness:{sha256:$liveness_sha,challenge:$liveness_challenge, + connection_id:$connection_id,session_id:$session_id, + destination_snapshot_sha256:$snapshot_sha,requested_at:$liveness_requested_at, + response_chunk_at:$liveness_response_chunk_at,received_at:$liveness_received_at}}, destination:{before:$destination_before,after:$destination_after, connection_count_delta:$destination_delta}, metrics:{source_service:(if $source=="legacy" then "legacy" else "slot" end), diff --git a/deploy/gcp/validate-deploy-evidence.py b/deploy/gcp/validate-deploy-evidence.py index 04696fc0a..41fc5c215 100755 --- a/deploy/gcp/validate-deploy-evidence.py +++ b/deploy/gcp/validate-deploy-evidence.py @@ -19,6 +19,7 @@ import argparse import datetime as dt +import hashlib import json import re import sys @@ -917,7 +918,7 @@ def validate_front_migration_transition(document: dict[str, Any], expected: str) challenge = text(field(proof, "challenge", "destination_proof"), "destination_proof.challenge") if not re.fullmatch(r"[0-9a-f]{32}", challenge): fail("destination_proof.challenge must be a 128-bit lowercase hex challenge") - text(field(proof, "connection_id", "destination_proof"), "destination_proof.connection_id") + connection_id = text(field(proof, "connection_id", "destination_proof"), "destination_proof.connection_id") session_id = text(field(proof, "session_id", "destination_proof"), "destination_proof.session_id") if re.fullmatch(r"[A-Za-z0-9._:-]{1,256}", session_id) is None: fail("destination_proof.session_id is invalid") @@ -953,6 +954,51 @@ def validate_front_migration_transition(document: dict[str, Any], expected: str) if proof_received - requested >= dt.timedelta(minutes=5): fail("destination proof exceeded the five-minute route propagation boundary") + liveness = obj( + field(proof, "post_snapshot_liveness", "destination_proof"), + "destination_proof.post_snapshot_liveness", + ) + sha(field(liveness, "sha256", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.sha256") + liveness_challenge = text( + field(liveness, "challenge", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.challenge", + ) + if not re.fullmatch(r"[0-9a-f]{32}", liveness_challenge): + fail("destination_proof.post_snapshot_liveness.challenge must be a 128-bit lowercase hex challenge") + exact( + text(field(liveness, "connection_id", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.connection_id"), + connection_id, + "destination_proof.post_snapshot_liveness.connection_id", + ) + exact( + text(field(liveness, "session_id", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.session_id"), + session_id, + "destination_proof.post_snapshot_liveness.session_id", + ) + liveness_snapshot_sha = sha( + field(liveness, "destination_snapshot_sha256", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.destination_snapshot_sha256", + ) + liveness_requested = timestamp( + field(liveness, "requested_at", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.requested_at", + ) + liveness_chunk = timestamp( + field(liveness, "response_chunk_at", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.response_chunk_at", + ) + liveness_received = timestamp( + field(liveness, "received_at", "destination_proof.post_snapshot_liveness"), + "destination_proof.post_snapshot_liveness.received_at", + ) + if not proof_received <= liveness_requested <= liveness_chunk <= liveness_received <= emitted: + fail("destination post-snapshot liveness timestamps are out of order") + if liveness_received - liveness_requested >= dt.timedelta(seconds=10): + fail("destination post-snapshot liveness exceeded ten seconds") + source = obj(field(document, "source", "root"), "source") before = validate_migration_snapshot(field(source, "before", "source"), "source.before", source_kind) after = validate_migration_snapshot(field(source, "after", "source"), "source.after", source_kind) @@ -1004,6 +1050,16 @@ def validate_front_migration_transition(document: dict[str, Any], expected: str) ) if destination_after["public_connections"] < 1 or destination_after["generation_connections"] < 1: fail("destination had no live connection after the journal-correlated golden proof") + canonical_destination = json.dumps( + destination_after, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + exact( + liveness_snapshot_sha, + hashlib.sha256(canonical_destination).hexdigest(), + "destination_proof.post_snapshot_liveness.destination_snapshot_sha256", + ) metrics = obj(field(document, "metrics", "root"), "metrics") exact( field(metrics, "source_service", "metrics"), From 941214cd577894b7eb4fee0dc310542c037c1971 Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Wed, 5 Aug 2026 07:38:57 -0700 Subject: [PATCH 5/5] fix: await migration evidence completion --- .../golden_migration_transition.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/cmd/subrouter-transport-observer/golden_migration_transition.go b/cmd/subrouter-transport-observer/golden_migration_transition.go index e75bcede0..c67af7334 100644 --- a/cmd/subrouter-transport-observer/golden_migration_transition.go +++ b/cmd/subrouter-transport-observer/golden_migration_transition.go @@ -250,6 +250,11 @@ func (r *goldenRunner) runMigrationTransitionWithProof( if err != nil || writePrivateFile(attemptLivenessProofPath, livenessProofFileData) != nil { return goldenActionSummary{}, nil, failGolden("migration_destination_liveness_proof_write_failed") } + select { + case <-ctx.Done(): + return goldenActionSummary{}, nil, ctx.Err() + case <-command.done: + } destinationSession = candidateSession break }