fix(harness): scope observer gap span suppression

commit: eed2010c969651de211378b9cafbe4addd8e76ab
author: brice · 2mo ago · 2026-08-02 10:57 UTC
parent: 7b0921e
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +59 -23

M cmd/vev-perf-harness/evidence.go +8 -5


@@ -200,21 +200,23 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) {
200	200		// and exact correlation domain may pair, and each pair's ticks must retain
201	201		// its own start-before-end ordering.
202	202		starts := map[string]traceRecord{}
203		-	observerGap := false
	203	+	discarded := map[string]struct{}{}
204	204		for _, r := range records {
205	205		if r.Component == "observer" && r.Kind == "transport_diagnostic" && !r.Valid {
206	206		// A bounded producer queue reports dropped marks explicitly. Any
207	207		// in-flight span may have lost either endpoint, so discard those
208		-	// partial samples; an end after the gap is not independently
209		-	// attributable to a missing start.
	208	+	// partial samples; only their exact keys may be missing an end.
	209	+	for key := range starts {
	210	+	discarded[key] = struct{}{}
	211	+	}
210	212		starts = map[string]traceRecord{}
211		-	observerGap = true
212	213		continue
213	214		}
214	215		for _, pair := range spanPairs {
215	216		key := fmt.Sprintf("%s/%s/%s/%d/%d/%d/%d", r.ProcessID, r.Component, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch)
216	217		if r.Kind == pair.start {
217	218		k := pair.name + "/" + key
	219	+	delete(discarded, k)
218	220		if _, exists := starts[k]; exists {
219	221		return nil, errors.New("duplicate process-local span start")
220	222		}

@@ -224,7 +226,8 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) {
224	226		k := pair.name + "/" + key
225	227		start, ok := starts[k]
226	228		if !ok {
227		-	if observerGap {
	229	+	if _, ok := discarded[k]; ok {
	230	+	delete(discarded, k)
228	231		continue
229	232		}
230	233		return nil, errors.New("span end without same-process start")

 		// and exact correlation domain may pair, and each pair's ticks must retain
 		// its own start-before-end ordering.
 		starts := map[string]traceRecord{}
-		observerGap := false
+		discarded := map[string]struct{}{}
 		for _, r := range records {
 			if r.Component == "observer" && r.Kind == "transport_diagnostic" && !r.Valid {
 				// A bounded producer queue reports dropped marks explicitly. Any
 				// in-flight span may have lost either endpoint, so discard those
-				// partial samples; an end after the gap is not independently
+				// partial samples; only their exact keys may be missing an end.
-				// attributable to a missing start.
+				for key := range starts {
+					discarded[key] = struct{}{}
+				}
 				starts = map[string]traceRecord{}
-				observerGap = true
 				continue
 			}
 			for _, pair := range spanPairs {
 				key := fmt.Sprintf("%s/%s/%s/%d/%d/%d/%d", r.ProcessID, r.Component, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch)
 				if r.Kind == pair.start {
 					k := pair.name + "/" + key
+					delete(discarded, k)
 					if _, exists := starts[k]; exists {
 						return nil, errors.New("duplicate process-local span start")
 					}
 					k := pair.name + "/" + key
 					start, ok := starts[k]
 					if !ok {
-						if observerGap {
+						if _, ok := discarded[k]; ok {
+							delete(discarded, k)
 							continue
 						}
 						return nil, errors.New("span end without same-process start")

M cmd/vev-perf-harness/evidence_test.go +51 -18


@@ -364,25 +364,58 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) {
364	364		}
365	365		}
366	366
367		-	func TestHarnessSkipsUnpairedEndAfterObserverGap(t *testing.T) {
368		-	path := filepath.Join(t.TempDir(), "trace.jsonl")
369		-	records := []traceRecord{
370		-	{ProcessID: "one", Component: "observer", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: "transport_diagnostic", Valid: false},
371		-	{ProcessID: "one", Component: "ipc", Scenario: "s", Run: 1, Sequence: 2, RequestID: 2, Epoch: 2, Kind: "adapter_receive_end", Tick: 20, Valid: true},
	367	+	func TestHarnessObserverGapSuppressesOnlyDiscardedSpanKeys(t *testing.T) {
	368	+	base := func(kind string, sequence, requestID, epoch uint64, tick int64) traceRecord {
	369	+	return traceRecord{ProcessID: "one", Component: "ipc", Scenario: "s", Run: 1, Sequence: sequence, RequestID: requestID, Epoch: epoch, Kind: kind, Tick: tick, Valid: true}
372	370		}
373		-	var lines []string
374		-	for _, record := range records {
375		-	lines = append(lines, mustJSON(record))
376		-	}
377		-	if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
378		-	t.Fatal(err)
379		-	}
380		-	spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})
381		-	if err != nil {
382		-	t.Fatalf("observer gap did not invalidate the dropped span: %v", err)
383		-	}
384		-	if len(spans) != 0 {
385		-	t.Fatalf("spans=%+v, want no sample after observer gap", spans)
	371	+	gap := traceRecord{ProcessID: "one", Component: "observer", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: "transport_diagnostic", Valid: false}
	372	+	for _, tc := range []struct {
	373	+	name string
	374	+	records []traceRecord
	375	+	wantErr bool
	376	+	}{
	377	+	{
	378	+	name: "discarded span end is suppressed",
	379	+	records: []traceRecord{
	380	+	base("adapter_receive_start", 2, 2, 2, 10),
	381	+	gap,
	382	+	base("adapter_receive_end", 2, 2, 2, 20),
	383	+	},
	384	+	},
	385	+	{
	386	+	name: "unrelated unmatched end remains an error",
	387	+	records: []traceRecord{
	388	+	base("adapter_receive_start", 2, 2, 2, 10),
	389	+	gap,
	390	+	base("adapter_receive_end", 2, 2, 2, 20),
	391	+	base("adapter_receive_end", 3, 3, 3, 30),
	392	+	},
	393	+	wantErr: true,
	394	+	},
	395	+	} {
	396	+	t.Run(tc.name, func(t *testing.T) {
	397	+	path := filepath.Join(t.TempDir(), "trace.jsonl")
	398	+	var lines []string
	399	+	for _, record := range tc.records {
	400	+	lines = append(lines, mustJSON(record))
	401	+	}
	402	+	if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
	403	+	t.Fatal(err)
	404	+	}
	405	+	spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})
	406	+	if tc.wantErr {
	407	+	if err == nil {
	408	+	t.Fatalf("unrelated unmatched end accepted: spans=%+v", spans)
	409	+	}
	410	+	return
	411	+	}
	412	+	if err != nil {
	413	+	t.Fatalf("discarded span end was rejected: %v", err)
	414	+	}
	415	+	if len(spans) != 0 {
	416	+	t.Fatalf("spans=%+v, want no sample after observer gap", spans)
	417	+	}
	418	+	})
386	419		}
387	420		}
388	421


@@ -364,25 +364,58 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) {
364		}	364		}
365		}	365		}
366			366
367	-	func TestHarnessSkipsUnpairedEndAfterObserverGap(t *testing.T) {	367	+	func TestHarnessObserverGapSuppressesOnlyDiscardedSpanKeys(t *testing.T) {
368	-	path := filepath.Join(t.TempDir(), "trace.jsonl")	368	+	base := func(kind string, sequence, requestID, epoch uint64, tick int64) traceRecord {
369	-	records := []traceRecord{	369	+	return traceRecord{ProcessID: "one", Component: "ipc", Scenario: "s", Run: 1, Sequence: sequence, RequestID: requestID, Epoch: epoch, Kind: kind, Tick: tick, Valid: true}
370	-	{ProcessID: "one", Component: "observer", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: "transport_diagnostic", Valid: false},
371	-	{ProcessID: "one", Component: "ipc", Scenario: "s", Run: 1, Sequence: 2, RequestID: 2, Epoch: 2, Kind: "adapter_receive_end", Tick: 20, Valid: true},
372		}	370		}
373	-	var lines []string	371	+	gap := traceRecord{ProcessID: "one", Component: "observer", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: "transport_diagnostic", Valid: false}
374	-	for _, record := range records {	372	+	for _, tc := range []struct {
375	-	lines = append(lines, mustJSON(record))	373	+	name string
376	-	}	374	+	records []traceRecord
377	-	if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {	375	+	wantErr bool
378	-	t.Fatal(err)	376	+	}{
379	-	}	377	+	{
380	-	spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})	378	+	name: "discarded span end is suppressed",
381	-	if err != nil {	379	+	records: []traceRecord{
382	-	t.Fatalf("observer gap did not invalidate the dropped span: %v", err)	380	+	base("adapter_receive_start", 2, 2, 2, 10),
383	-	}	381	+	gap,
384	-	if len(spans) != 0 {	382	+	base("adapter_receive_end", 2, 2, 2, 20),
385	-	t.Fatalf("spans=%+v, want no sample after observer gap", spans)	383	+	},
			384	+	},
			385	+	{
			386	+	name: "unrelated unmatched end remains an error",
			387	+	records: []traceRecord{
			388	+	base("adapter_receive_start", 2, 2, 2, 10),
			389	+	gap,
			390	+	base("adapter_receive_end", 2, 2, 2, 20),
			391	+	base("adapter_receive_end", 3, 3, 3, 30),
			392	+	},
			393	+	wantErr: true,
			394	+	},
			395	+	} {
			396	+	t.Run(tc.name, func(t *testing.T) {
			397	+	path := filepath.Join(t.TempDir(), "trace.jsonl")
			398	+	var lines []string
			399	+	for _, record := range tc.records {
			400	+	lines = append(lines, mustJSON(record))
			401	+	}
			402	+	if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
			403	+	t.Fatal(err)
			404	+	}
			405	+	spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})
			406	+	if tc.wantErr {
			407	+	if err == nil {
			408	+	t.Fatalf("unrelated unmatched end accepted: spans=%+v", spans)
			409	+	}
			410	+	return
			411	+	}
			412	+	if err != nil {
			413	+	t.Fatalf("discarded span end was rejected: %v", err)
			414	+	}
			415	+	if len(spans) != 0 {
			416	+	t.Fatalf("spans=%+v, want no sample after observer gap", spans)
			417	+	}
			418	+	})
386		}	419		}
387		}	420		}
388			421