fix(obs): validate concurrent trace ordering
- commit
- 0021186a3a4ac09b3e10969fb34734d90971df0b
- author
- brice · · 2026-07-13 00:37 UTC
- parent
- 720c52d
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +37 -13
M cmd/vev-perf-harness/main.go +7 -13
| @@ -1427,22 +1427,16 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) { | |||
| 1427 | 1427 | return nil, err | |
| 1428 | 1428 | } | |
| 1429 | 1429 | _ = f.Close() | |
| 1430 | - | // Spans may overlap: an end for an older operation can legitimately be | |
| 1431 | - | // serialized after a newer operation's start. New correlation IDs still | |
| 1432 | - | // must be allocated monotonically within the process. | |
| 1433 | - | lastSequence := uint64(0) | |
| 1434 | - | seenSequences := map[uint64]bool{} | |
| 1430 | + | // A process shares one observer across concurrent components. Correlation | |
| 1431 | + | // IDs are allocated before a goroutine reaches its first mark, so a later | |
| 1432 | + | // ID can be serialized before an earlier one. Their first appearance is | |
| 1433 | + | // consequently not an ordering contract. Only marks in the same component | |
| 1434 | + | // and exact correlation domain may pair, and each pair's ticks must retain | |
| 1435 | + | // its own start-before-end ordering. | |
| 1435 | 1436 | starts := map[string]traceRecord{} | |
| 1436 | 1437 | for _, r := range records { | |
| 1437 | - | if !seenSequences[r.Sequence] { | |
| 1438 | - | if r.Sequence < lastSequence { | |
| 1439 | - | return nil, errors.New("nonmonotonic same-process sequence") | |
| 1440 | - | } | |
| 1441 | - | seenSequences[r.Sequence] = true | |
| 1442 | - | lastSequence = r.Sequence | |
| 1443 | - | } | |
| 1444 | 1438 | for _, pair := range spanPairs { | |
| 1445 | - | key := fmt.Sprintf("%s/%s/%d/%d/%d/%d", r.ProcessID, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch) | |
| 1439 | + | key := fmt.Sprintf("%s/%s/%s/%d/%d/%d/%d", r.ProcessID, r.Component, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch) | |
| 1446 | 1440 | if r.Kind == pair.start { | |
| 1447 | 1441 | k := pair.name + "/" + key | |
| 1448 | 1442 | if _, exists := starts[k]; exists { | |
| @@ -1427,22 +1427,16 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) { | |||||
| 1427 | return nil, err | 1427 | return nil, err | ||
| 1428 | } | 1428 | } | ||
| 1429 | _ = f.Close() | 1429 | _ = f.Close() | ||
| 1430 | - | // Spans may overlap: an end for an older operation can legitimately be | 1430 | + | // A process shares one observer across concurrent components. Correlation |
| 1431 | - | // serialized after a newer operation's start. New correlation IDs still | 1431 | + | // IDs are allocated before a goroutine reaches its first mark, so a later |
| 1432 | - | // must be allocated monotonically within the process. | 1432 | + | // ID can be serialized before an earlier one. Their first appearance is |
| 1433 | - | lastSequence := uint64(0) | 1433 | + | // consequently not an ordering contract. Only marks in the same component |
| 1434 | - | seenSequences := map[uint64]bool{} | 1434 | + | // and exact correlation domain may pair, and each pair's ticks must retain |
| 1435 | + | // its own start-before-end ordering. | |||
| 1435 | starts := map[string]traceRecord{} | 1436 | starts := map[string]traceRecord{} | ||
| 1436 | for _, r := range records { | 1437 | for _, r := range records { | ||
| 1437 | - | if !seenSequences[r.Sequence] { | |||
| 1438 | - | if r.Sequence < lastSequence { | |||
| 1439 | - | return nil, errors.New("nonmonotonic same-process sequence") | |||
| 1440 | - | } | |||
| 1441 | - | seenSequences[r.Sequence] = true | |||
| 1442 | - | lastSequence = r.Sequence | |||
| 1443 | - | } | |||
| 1444 | for _, pair := range spanPairs { | 1438 | for _, pair := range spanPairs { | ||
| 1445 | - | key := fmt.Sprintf("%s/%s/%d/%d/%d/%d", r.ProcessID, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch) | 1439 | + | key := fmt.Sprintf("%s/%s/%s/%d/%d/%d/%d", r.ProcessID, r.Component, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch) |
| 1446 | if r.Kind == pair.start { | 1440 | if r.Kind == pair.start { | ||
| 1447 | k := pair.name + "/" + key | 1441 | k := pair.name + "/" + key | ||
| 1448 | if _, exists := starts[k]; exists { | 1442 | if _, exists := starts[k]; exists { | ||
M cmd/vev-perf-harness/main_test.go +30 -0
| @@ -1036,6 +1036,34 @@ func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) { | |||
| 1036 | 1036 | } | |
| 1037 | 1037 | } | |
| 1038 | 1038 | } | |
| 1039 | + | ||
| 1040 | + | // boundedLocalConcurrentTrace is an excerpt from the second run of the | |
| 1041 | + | // bounded local public-CLI smoke. Receive sequence 3 is in flight while the | |
| 1042 | + | // same process serializes sequence 4; shared-process records therefore cannot | |
| 1043 | + | // have a global sequence ordering requirement. | |
| 1044 | + | const boundedLocalConcurrentTrace = `{"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":3,"request_id":3,"epoch":3,"kind":"adapter_receive_start","tick":1783902621495284782,"bytes":0,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |
| 1045 | + | {"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":4,"request_id":4,"epoch":4,"kind":"adapter_send_start","tick":1783902621495364552,"bytes":96,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |
| 1046 | + | {"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":4,"request_id":4,"epoch":4,"kind":"adapter_send_end","tick":1783902621495386452,"bytes":96,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |
| 1047 | + | {"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":3,"request_id":3,"epoch":3,"kind":"adapter_receive_end","tick":1783902621495463332,"bytes":0,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |
| 1048 | + | ` | |
| 1049 | + | ||
| 1050 | + | func TestHarnessAcceptsConcurrentProductionTraceInterleaving(t *testing.T) { | |
| 1051 | + | path := filepath.Join(t.TempDir(), "local-client.jsonl") | |
| 1052 | + | if err := os.WriteFile(path, []byte(boundedLocalConcurrentTrace), 0o600); err != nil { | |
| 1053 | + | t.Fatal(err) | |
| 1054 | + | } | |
| 1055 | + | spans, err := mergeProcessTraces([]processMapping{{ | |
| 1056 | + | ProcessID: "1x4-idle-local-r002-client-02", ClockDomain: "1x4-idle-local-r002-client-02", TracePath: path, | |
| 1057 | + | Scenario: "1x4-idle-local", Run: 2, | |
| 1058 | + | }}) | |
| 1059 | + | if err != nil { | |
| 1060 | + | t.Fatalf("valid concurrent production trace rejected: %v", err) | |
| 1061 | + | } | |
| 1062 | + | if len(spans) != 2 || spans[0].Component != "ipc" || spans[0].Name != "adapter_send_duration" || spans[0].Samples[0] != 21900 || spans[1].Component != "ipc" || spans[1].Name != "adapter_receive_duration" || spans[1].Samples[0] != 178550 { | |
| 1063 | + | t.Fatalf("spans=%+v", spans) | |
| 1064 | + | } | |
| 1065 | + | } | |
| 1066 | + | ||
| 1039 | 1067 | func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |
| 1040 | 1068 | write := func(t *testing.T, records ...traceRecord) processMapping { | |
| 1041 | 1069 | t.Helper() | |
| @@ -1058,6 +1086,8 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||
| 1058 | 1086 | {base("diff_end", 1)}, | |
| 1059 | 1087 | {base("diff_start", 2), base("diff_end", 1)}, | |
| 1060 | 1088 | {base("diff_start", 1)}, | |
| 1089 | + | {base("diff_start", 1), base("diff_start", 2)}, | |
| 1090 | + | {base("diff_start", 1), func() traceRecord { r := base("diff_end", 2); r.Component = "ipc"; return r }()}, | |
| 1061 | 1091 | {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}, | |
| 1062 | 1092 | {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()}, | |
| 1063 | 1093 | {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()}, | |
| @@ -1036,6 +1036,34 @@ func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) { | |||||
| 1036 | } | 1036 | } | ||
| 1037 | } | 1037 | } | ||
| 1038 | } | 1038 | } | ||
| 1039 | + | ||||
| 1040 | + | // boundedLocalConcurrentTrace is an excerpt from the second run of the | |||
| 1041 | + | // bounded local public-CLI smoke. Receive sequence 3 is in flight while the | |||
| 1042 | + | // same process serializes sequence 4; shared-process records therefore cannot | |||
| 1043 | + | // have a global sequence ordering requirement. | |||
| 1044 | + | const boundedLocalConcurrentTrace = `{"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":3,"request_id":3,"epoch":3,"kind":"adapter_receive_start","tick":1783902621495284782,"bytes":0,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |||
| 1045 | + | {"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":4,"request_id":4,"epoch":4,"kind":"adapter_send_start","tick":1783902621495364552,"bytes":96,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |||
| 1046 | + | {"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":4,"request_id":4,"epoch":4,"kind":"adapter_send_end","tick":1783902621495386452,"bytes":96,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |||
| 1047 | + | {"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":3,"request_id":3,"epoch":3,"kind":"adapter_receive_end","tick":1783902621495463332,"bytes":0,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true} | |||
| 1048 | + | ` | |||
| 1049 | + | ||||
| 1050 | + | func TestHarnessAcceptsConcurrentProductionTraceInterleaving(t *testing.T) { | |||
| 1051 | + | path := filepath.Join(t.TempDir(), "local-client.jsonl") | |||
| 1052 | + | if err := os.WriteFile(path, []byte(boundedLocalConcurrentTrace), 0o600); err != nil { | |||
| 1053 | + | t.Fatal(err) | |||
| 1054 | + | } | |||
| 1055 | + | spans, err := mergeProcessTraces([]processMapping{{ | |||
| 1056 | + | ProcessID: "1x4-idle-local-r002-client-02", ClockDomain: "1x4-idle-local-r002-client-02", TracePath: path, | |||
| 1057 | + | Scenario: "1x4-idle-local", Run: 2, | |||
| 1058 | + | }}) | |||
| 1059 | + | if err != nil { | |||
| 1060 | + | t.Fatalf("valid concurrent production trace rejected: %v", err) | |||
| 1061 | + | } | |||
| 1062 | + | if len(spans) != 2 || spans[0].Component != "ipc" || spans[0].Name != "adapter_send_duration" || spans[0].Samples[0] != 21900 || spans[1].Component != "ipc" || spans[1].Name != "adapter_receive_duration" || spans[1].Samples[0] != 178550 { | |||
| 1063 | + | t.Fatalf("spans=%+v", spans) | |||
| 1064 | + | } | |||
| 1065 | + | } | |||
| 1066 | + | ||||
| 1039 | func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | 1067 | func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | ||
| 1040 | write := func(t *testing.T, records ...traceRecord) processMapping { | 1068 | write := func(t *testing.T, records ...traceRecord) processMapping { | ||
| 1041 | t.Helper() | 1069 | t.Helper() | ||
| @@ -1058,6 +1086,8 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||||
| 1058 | {base("diff_end", 1)}, | 1086 | {base("diff_end", 1)}, | ||
| 1059 | {base("diff_start", 2), base("diff_end", 1)}, | 1087 | {base("diff_start", 2), base("diff_end", 1)}, | ||
| 1060 | {base("diff_start", 1)}, | 1088 | {base("diff_start", 1)}, | ||
| 1089 | + | {base("diff_start", 1), base("diff_start", 2)}, | |||
| 1090 | + | {base("diff_start", 1), func() traceRecord { r := base("diff_end", 2); r.Component = "ipc"; return r }()}, | |||
| 1061 | {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}, | 1091 | {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}, | ||
| 1062 | {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()}, | 1092 | {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()}, | ||
| 1063 | {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()}, | 1093 | {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()}, | ||