[review-fix 2] Retry one-shots under backpressure

- Retain one-shot contexts until the collector processing channel accepts them, so full-channel drops remain retryable within the selection window.
- Acknowledge accepted one-shots with deadline identity checks to avoid deleting a newer same-path window.
- Cover store acknowledgment, cross-window races, and processing-channel backpressure.

Source: review feedback
Validation: Bazel race tests for collector and path-test store

Author: AlexandreYang

comp/networkpath/npcollector/impl/npcollector.go

@@ -595,6 +595,7 @@ func (s *npCollectorImpl) flush() {
 		s.logger.Tracef("flushed ptConf %s:%d", ptConf.Pathtest.Hostname, ptConf.Pathtest.Port)
 		select {
 		case s.pathtestProcessingChan <- ptConf:
+			s.pathtestStore.AcknowledgeOneShot(ptConf.Pathtest)
 			_ = s.statsdClient.Incr(common.NetworkPathCollectorMetricPrefix+"flush.pathtest_processed", []string{}, 1)
 		default:
 			_ = s.statsdClient.Incr(common.NetworkPathCollectorMetricPrefix+"flush.pathtest_dropped", []string{"reason:processing_chan_full"}, 1)

comp/networkpath/npcollector/impl/npcollector_test.go

@@ -1678,6 +1678,31 @@ func Test_npCollectorImpl_flush(t *testing.T) {
 	assert.Equal(t, 2, len(npCollector.pathtestProcessingChan))
 }
 
+func TestNpCollectorFlushRetriesOneShotAfterProcessingChannelBackpressure(t *testing.T) {
+	_, collector := newTestNpCollector(t, map[string]any{
+		"network_path.connections_monitoring.baseline_tests.enabled": true,
+		"network_path.collector.processing_chan_size":                1,
+	}, &teststatsd.Client{}, nil)
+	deadline := time.Now().Add(time.Minute)
+	collector.pathtestStore.Add(&common.Pathtest{
+		Hostname:           "baseline",
+		Origin:             payload.PathOriginNetworkTraffic,
+		OneShot:            true,
+		ExecutionDeadline:  deadline,
+		DynamicTestProfile: payload.DynamicTestProfileBaseline,
+	})
+	collector.pathtestProcessingChan <- &pathteststore.PathtestContext{}
+
+	collector.flush()
+	assert.Equal(t, 1, collector.pathtestStore.GetContextsCount(), "full channel must retain the one-shot")
+	<-collector.pathtestProcessingChan
+
+	collector.flush()
+	require.Len(t, collector.pathtestProcessingChan, 1)
+	assert.Equal(t, "baseline", (<-collector.pathtestProcessingChan).Pathtest.Hostname)
+	assert.Zero(t, collector.pathtestStore.GetContextsCount())
+}
+
 func Test_npCollectorImpl_flushLoop(t *testing.T) {
 	// GIVEN
 	agentConfigs := map[string]any{

comp/networkpath/npcollector/impl/pathteststore/pathteststore.go

@@ -171,18 +171,18 @@ func (f *Store) Flush() []*PathtestContext {
 		if ptConfigCtx.nextRun.After(now) || !f.rateLimiter.AllowN(now, 1) {
 			continue
 		}
+		if ptConfigCtx.Pathtest.OneShot {
+			// Keep one-shots until the collector acknowledges that its processing
+			// channel accepted the snapshot. A full channel must leave the selection
+			// available for the next flush attempt.
+			pathtestsToFlush = append(pathtestsToFlush, ptConfigCtx.snapshot())
+			continue
+		}
 		if !ptConfigCtx.lastFlushTime.IsZero() {
 			ptConfigCtx.lastFlushInterval = now.Sub(ptConfigCtx.lastFlushTime)
 		}
 		ptConfigCtx.lastFlushTime = now
 		pathtestsToFlush = append(pathtestsToFlush, ptConfigCtx.snapshot())
-		if ptConfigCtx.Pathtest.OneShot {
-			if ptConfigCtx.Pathtest.DynamicTestProfile == payload.DynamicTestProfileBaseline {
-				f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_dispatched", []string{}, 1) //nolint:errcheck
-			}
-			delete(f.contexts, key)
-			continue
-		}
 		ptConfigCtx.nextRun = ptConfigCtx.nextRun.Add(f.config.Interval)
 	}
 
@@ -191,6 +191,28 @@ func (f *Store) Flush() []*PathtestContext {
 	return pathtestsToFlush
 }
 
+// AcknowledgeOneShot removes a one-shot after the collector processing channel
+// accepts it. The deadline check prevents a late acknowledgment from removing a
+// same-path selection admitted for a newer window.
+func (f *Store) AcknowledgeOneShot(pathtest *common.Pathtest) {
+	if !pathtest.OneShot {
+		return
+	}
+
+	f.contextsMutex.Lock()
+	defer f.contextsMutex.Unlock()
+
+	hash := pathtest.GetHash()
+	pathtestCtx, ok := f.contexts[hash]
+	if !ok || !pathtestCtx.Pathtest.OneShot || !pathtestCtx.Pathtest.ExecutionDeadline.Equal(pathtest.ExecutionDeadline) {
+		return
+	}
+	delete(f.contexts, hash)
+	if pathtestCtx.Pathtest.DynamicTestProfile == payload.DynamicTestProfileBaseline {
+		f.statsdClient.Incr(networkPathStoreMetricPrefix+"baseline_dispatched", []string{}, 1) //nolint:errcheck
+	}
+}
+
 // Add new pathtest
 func (f *Store) Add(pathtestToAdd *common.Pathtest) {
 	f.logger.Tracef("Add new Pathtest: %+v", pathtestToAdd)

comp/networkpath/npcollector/impl/pathteststore/pathteststore_test.go

@@ -207,6 +207,8 @@ func TestPathtestStoreOneShotDispatchesOnceAndDeletes(t *testing.T) {
 	flushed := store.Flush()
 	require.Len(t, flushed, 1)
 	assert.True(t, flushed[0].Pathtest.OneShot)
+	assert.Equal(t, 1, store.GetContextsCount(), "one-shot must remain pending until channel admission")
+	store.AcknowledgeOneShot(flushed[0].Pathtest)
 	assert.Zero(t, store.GetContextsCount())
 	assert.Empty(t, store.Flush())
 	assert.Equal(t, int64(1), stats.GetCountSummaries()[networkPathStoreMetricPrefix+"baseline_dispatched"].Sum)
@@ -236,12 +238,32 @@ func TestPathtestStoreReplacesExpiredOneShotBeforeFlush(t *testing.T) {
 	replacement := store.contexts[(&common.Pathtest{Hostname: "baseline"}).GetHash()]
 	require.NotNil(t, replacement)
 	assert.Equal(t, newDeadline, replacement.runUntil)
-	assert.Len(t, store.Flush(), 1, "the next window's replacement must remain dispatchable")
+	flushed := store.Flush()
+	require.Len(t, flushed, 1, "the next window's replacement must remain dispatchable")
+	store.AcknowledgeOneShot(flushed[0].Pathtest)
 	assert.Zero(t, store.GetContextsCount())
 	assert.Equal(t, int64(1), stats.GetCountSummaries()[networkPathStoreMetricPrefix+"baseline_expired"].Sum)
 	assert.Equal(t, int64(1), stats.GetCountSummaries()[networkPathStoreMetricPrefix+"baseline_dispatched"].Sum)
 }
 
+func TestPathtestStoreOneShotAcknowledgmentDoesNotRemoveNewWindow(t *testing.T) {
+	now := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC)
+	store := NewPathtestStore(Config{ContextsLimit: 10}, logmock.New(t), &statsd.NoOpClient{}, func() time.Time { return now })
+	store.Add(&common.Pathtest{Hostname: "baseline", OneShot: true, ExecutionDeadline: now.Add(30 * time.Minute)})
+	flushed := store.Flush()
+	require.Len(t, flushed, 1)
+	oldWindow := flushed[0].Pathtest
+	now = now.Add(30 * time.Minute)
+	newDeadline := now.Add(30 * time.Minute)
+	store.Add(&common.Pathtest{Hostname: "baseline", OneShot: true, ExecutionDeadline: newDeadline})
+
+	store.AcknowledgeOneShot(oldWindow)
+
+	context := store.contexts[oldWindow.GetHash()]
+	require.NotNil(t, context)
+	assert.Equal(t, newDeadline, context.runUntil)
+}
+
 func TestPathtestStoreRecurringDispatchesAtTTLBoundary(t *testing.T) {
 	now := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC)
 	store := NewPathtestStore(Config{