fix(seahorse): add mutex for assembler lazy initialization
Issue #5 from PR review: The check-then-create pattern for e.assembler was a data race when multiple goroutines called Assemble() concurrently: if e.assembler == nil { e.assembler = &Assembler{...} } Changes: - Add assemblerMu sync.Mutex to Engine struct - Add initAssemblerOnce() using double-checked locking (same pattern as initCompactionOnce) - Add TestAssemblerLazyInitRace to verify thread-safety
This commit is contained in:
parent
a2f72f45e6
commit
971b992e78
3 changed files with 59 additions and 3 deletions
2
.gitignore
vendored
2
.gitignore
vendored
|
|
@ -67,3 +67,5 @@ web/backend/dist/*
|
||||||
.claude/
|
.claude/
|
||||||
|
|
||||||
docker/data
|
docker/data
|
||||||
|
|
||||||
|
.omc/
|
||||||
|
|
|
||||||
|
|
@ -58,6 +58,7 @@ type Engine struct {
|
||||||
compaction *CompactionEngine
|
compaction *CompactionEngine
|
||||||
compactionMu sync.Mutex
|
compactionMu sync.Mutex
|
||||||
assembler *Assembler
|
assembler *Assembler
|
||||||
|
assemblerMu sync.Mutex
|
||||||
retrieval *RetrievalEngine
|
retrieval *RetrievalEngine
|
||||||
config Config
|
config Config
|
||||||
complete CompleteFn
|
complete CompleteFn
|
||||||
|
|
@ -307,9 +308,7 @@ func (e *Engine) Assemble(ctx context.Context, sessionKey string, input Assemble
|
||||||
return nil, fmt.Errorf("get conversation: %w", err)
|
return nil, fmt.Errorf("get conversation: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if e.assembler == nil {
|
e.initAssemblerOnce()
|
||||||
e.assembler = &Assembler{store: e.store, config: e.config}
|
|
||||||
}
|
|
||||||
return e.assembler.Assemble(ctx, conv.ConversationID, input)
|
return e.assembler.Assemble(ctx, conv.ConversationID, input)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -362,6 +361,17 @@ func (e *Engine) initCompactionOnce() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// initAssemblerOnce lazily initializes the assembler.
|
||||||
|
func (e *Engine) initAssemblerOnce() {
|
||||||
|
if e.assembler == nil {
|
||||||
|
e.assemblerMu.Lock()
|
||||||
|
defer e.assemblerMu.Unlock()
|
||||||
|
if e.assembler == nil {
|
||||||
|
e.assembler = &Assembler{store: e.store, config: e.config}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// IngestMessages is an alias for Ingest.
|
// IngestMessages is an alias for Ingest.
|
||||||
func (e *Engine) IngestMessages(ctx context.Context, sessionKey string, messages []Message) (*IngestResult, error) {
|
func (e *Engine) IngestMessages(ctx context.Context, sessionKey string, messages []Message) (*IngestResult, error) {
|
||||||
return e.Ingest(ctx, sessionKey, messages)
|
return e.Ingest(ctx, sessionKey, messages)
|
||||||
|
|
|
||||||
|
|
@ -1329,3 +1329,47 @@ func TestBootstrapAnchorWithDuplicateContent_Simple(t *testing.T) {
|
||||||
t.Errorf("last message content = %q, want 'D'", lastMsg.Content)
|
t.Errorf("last message content = %q, want 'D'", lastMsg.Content)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --- Assembler lazy init race detection ---
|
||||||
|
|
||||||
|
func TestAssemblerLazyInitRace(t *testing.T) {
|
||||||
|
// This test verifies that Assemble() lazy initialization of e.assembler
|
||||||
|
// is thread-safe. The original code has a data race:
|
||||||
|
// if e.assembler == nil {
|
||||||
|
// e.assembler = &Assembler{...}
|
||||||
|
// }
|
||||||
|
|
||||||
|
// Run multiple iterations to increase chance of catching race
|
||||||
|
for i := 0; i < 30; i++ {
|
||||||
|
// Create fresh engine with nil assembler
|
||||||
|
e := newTestEngine(t)
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
sessionKey := fmt.Sprintf("race-test-%d", i)
|
||||||
|
|
||||||
|
// Add message first (avoid SQLite concurrency issues)
|
||||||
|
_, err := e.Ingest(ctx, sessionKey, []Message{
|
||||||
|
{Role: "user", Content: "hello", TokenCount: 5},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Ingest: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Use a barrier to ensure all goroutines start at the same time
|
||||||
|
start := make(chan struct{})
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
|
||||||
|
for j := 0; j < 20; j++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
<-start // Wait for all goroutines to be ready
|
||||||
|
e.Assemble(ctx, sessionKey, AssembleInput{Budget: 1000})
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Start all goroutines simultaneously
|
||||||
|
close(start)
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue