eric/weatherreporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Make prompt debug creation concurrency safe

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-02 13:11:41 +00:00
parent e0229d9c90
commit 1716702c99
4 changed files with 453 additions and 740 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
internal/app/comparison_execution_test.go
View File

@@ -26,7 +26,7 @@ func TestExecuteComparisonProfilesRunsOrderedProfilesConcurrently(t *testing.T)
Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor,
})
}()
waitForProfileStarts(t, executor, profiles)
waitForProfileStarts(t, executor, profiles, results)
if executor.maximumInFlight() < 2 {
t.Fatalf("maximum in-flight executions = %d, want overlap", executor.maximumInFlight())
}
@@ -64,7 +64,7 @@ func TestExecuteComparisonProfilesContinuesAfterProfileFailure(t *testing.T) {
Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor,
})
}()
waitForProfileStarts(t, executor, profiles)
waitForProfileStarts(t, executor, profiles, results)
for _, profile := range profiles {
executor.release(profile.ProfileID)
}
@@ -90,7 +90,7 @@ func TestExecuteComparisonProfilesPropagatesCancellationAndJoins(t *testing.T) {
Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", Executor: executor,
})
}()
waitForProfileStarts(t, executor, profiles)
waitForProfileStarts(t, executor, profiles, results)
cancel()
result := <-results
if !result.Canceled || executor.inFlightCount() != 0 {
@@ -120,7 +120,7 @@ func TestExecuteComparisonProfilesUsesDistinctDeterministicDebugReferences(t *te
Prepared: prepared, Inspection: comparisonInspection(prompt, profiles), ComparisonID: "comparison_daily", DebugWriter: debugWriter, Executor: executor,
})
}()
waitForProfileStarts(t, executor, profiles)
waitForProfileStarts(t, executor, profiles, results)
for _, profile := range profiles {
executor.release(profile.ProfileID)
}
@@ -142,13 +142,14 @@ func TestExecuteComparisonProfilesUsesDistinctDeterministicDebugReferences(t *te
}
type barrierExecutor struct {
mu sync.Mutex
started chan string
releases map[string]chan struct{}
requests map[string]promptexec.ExecuteRequest
errors map[string]error
inFlight int
maximum int
mu sync.Mutex
started chan string
callbackFailures chan error
releases map[string]chan struct{}
requests map[string]promptexec.ExecuteRequest
errors map[string]error
inFlight int
maximum int
}
func newBarrierExecutor(profiles []ComparisonProfileInspection) *barrierExecutor {
@@ -157,7 +158,7 @@ func newBarrierExecutor(profiles []ComparisonProfileInspection) *barrierExecutor
releases[profile.ProfileID] = make(chan struct{})
}
return &barrierExecutor{
started: make(chan string, len(profiles)), releases: releases,
started: make(chan string, len(profiles)), callbackFailures: make(chan error, len(profiles)), releases: releases,
requests: make(map[string]promptexec.ExecuteRequest, len(profiles)), errors: map[string]error{},
}
}
@@ -173,6 +174,7 @@ func (e *barrierExecutor) InspectProfile(context.Context, string) (promptexec.Pr
func (e *barrierExecutor) Execute(ctx context.Context, req promptexec.ExecuteRequest, callback promptexec.PreparationCallback) (*promptexec.Execution, error) {
stamp := time.Date(2026, 5, 29, 15, 0, 0, 0, time.UTC)
if err := callback(promptexec.Preparation{PromptID: req.PromptID, PromptVersion: req.PromptVersion, PromptHash: "prompt-hash", ProfileID: req.ProfileID, BackendID: "backend-" + req.ProfileID, ModelName: "model-" + req.ProfileID, StartedAt: stamp, EndedAt: stamp}, nil); err != nil {
e.callbackFailures <- err
return nil, err
}
e.mu.Lock()
@@ -224,6 +226,16 @@ func (e *barrierExecutor) release(profileID string) {
close(e.releases[profileID])
}
func (e *barrierExecutor) releaseAll() {
for _, release := range e.releases {
select {
case <-release:
default:
close(release)
}
}
}
func (e *barrierExecutor) maximumInFlight() int {
e.mu.Lock()
defer e.mu.Unlock()
@@ -236,11 +248,28 @@ func (e *barrierExecutor) inFlightCount() int {
return e.inFlight
}
func waitForProfileStarts(t *testing.T, executor *barrierExecutor, profiles []ComparisonProfileInspection) {
func waitForProfileStarts(t *testing.T, executor *barrierExecutor, profiles []ComparisonProfileInspection, results <-chan comparisonExecutionResult) {
t.Helper()
timeout := time.NewTimer(5 * time.Second)
defer timeout.Stop()
seen := map[string]struct{}{}
for range profiles {
profileID := <-executor.started
var profileID string
select {
case profileID = <-executor.started:
case err := <-executor.callbackFailures:
executor.releaseAll()
select {
case result := <-results:
t.Fatalf("comparison profile preparation failed before executor entry: %v; result: %#v", err, result)
case <-timeout.C:
t.Fatalf("comparison profile preparation failed before executor entry: %v; comparison did not finish", err)
}
case result := <-results:
t.Fatalf("comparison completed before all profiles started: %#v", result)
case <-timeout.C:
t.Fatal("timed out waiting for comparison profile starts")
}
if _, duplicate := seen[profileID]; duplicate {
t.Fatalf("duplicate execution start for %q", profileID)
}