Skip to content

Correlate golden migration proof by session #176

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lawrencecchen merged 5 commits into from
Aug 5, 2026
Merged

Correlate golden migration proof by session #176

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6508e18
test: cover concurrent destination drains
lawrencecchen Aug 5, 2026
1e19407
fix: correlate migration proof by session
lawrencecchen Aug 5, 2026
e0d3b38
test: require post-snapshot migration liveness
lawrencecchen Aug 5, 2026
5a62893
fix: prove destination session stays live
lawrencecchen Aug 5, 2026
941214c
fix: await migration evidence completion
lawrencecchen Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cmd/subrouter-transport-observer/golden.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
143 changes: 139 additions & 4 deletions cmd/subrouter-transport-observer/golden_acceptance_regression_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,15 +30,21 @@ 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)
}

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")
}
Expand All @@ -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)
Expand All @@ -66,6 +75,125 @@ 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
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)
}
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)
}

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 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 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),
Expand Down Expand Up @@ -931,8 +1059,15 @@ 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",
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)},
Expand Down
63 changes: 52 additions & 11 deletions cmd/subrouter-transport-observer/golden_migration_evidence.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,14 +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"`
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 {
Expand Down Expand Up @@ -466,9 +479,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(
Expand All @@ -485,9 +497,23 @@ 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")
}
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
}
Expand All @@ -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" {
Expand Down
Loading
Loading