From e7d4a9c822409e166ae66540d87455ef47e44865 Mon Sep 17 00:00:00 2001 From: Steve Yegge Date: Wed, 15 Oct 2025 19:13:27 -0700 Subject: [PATCH] Implement CreateIssues batch API (bd-240, bd-244) - Add CreateIssues method to Storage interface - Implement SQLiteStorage.CreateIssues with atomic ID range reservation - Single transaction for N issues (5-10x speedup expected) - Set timestamps before validation to match CreateIssue behavior - All tests passing --- .beads/issues.jsonl | 4 +- internal/storage/sqlite/sqlite.go | 173 ++++++++++++++++++++++++++++++ internal/storage/storage.go | 1 + 3 files changed, 176 insertions(+), 2 deletions(-) diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index a8a61c7a..a8937099 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -155,11 +155,11 @@ {"id":"bd-238","title":"Fix UpdateIssue to auto-manage closed_at on status changes","description":"Modify internal/storage/sqlite/sqlite.go UpdateIssue to automatically manage closed_at when status changes.\n\nWhen status changes TO closed: set closed_at = now()\nWhen status changes FROM closed: set closed_at = nil and emit EventReopened","design":"```go\nfunc (s *SQLiteStorage) UpdateIssue(ctx context.Context, id string, updates map[string]interface{}, actor string) error {\n // ... existing validation ...\n\n // Smart closed_at management based on status changes\n if statusVal, ok := updates[\"status\"]; ok {\n newStatus := statusVal.(string)\n\n if newStatus == string(types.StatusClosed) {\n // Changing to closed: ensure closed_at is set\n if _, hasClosedAt := updates[\"closed_at\"]; !hasClosedAt {\n updates[\"closed_at\"] = time.Now()\n }\n } else {\n // Changing from closed to something else: clear closed_at\n if oldIssue.Status == types.StatusClosed {\n updates[\"closed_at\"] = nil\n eventType = types.EventReopened\n }\n }\n }\n\n // ... rest of existing code ...\n}\n```","status":"closed","priority":1,"issue_type":"task","created_at":"2025-10-15T14:21:05.817245-07:00","updated_at":"2025-10-15T16:27:22.060053-07:00","closed_at":"2025-10-15T16:27:22.060053-07:00","dependencies":[{"issue_id":"bd-238","depends_on_id":"bd-224","type":"parent-child","created_at":"2025-10-15T14:21:05.819159-07:00","created_by":"stevey"},{"issue_id":"bd-238","depends_on_id":"bd-239","type":"blocks","created_at":"2025-10-15T14:22:35.448877-07:00","created_by":"stevey"}]} {"id":"bd-239","title":"Fix import.go to normalize closed_at before creating issues","description":"Modify cmd/bd/import.go to enforce closed_at invariant before calling CreateIssue/CreateIssues.\n\nNormalize data: if status=closed, ensure closed_at is set; if status!=closed, ensure closed_at is nil.","design":"```go\nif _, ok := rawData[\"status\"]; ok {\n updates[\"status\"] = issue.Status\n\n // Enforce closed_at invariant\n if issue.Status == types.StatusClosed {\n // Status is closed: ensure closed_at is set\n if issue.ClosedAt == nil {\n now := time.Now()\n updates[\"closed_at\"] = now\n } else {\n updates[\"closed_at\"] = *issue.ClosedAt\n }\n } else {\n // Status is not closed: ensure closed_at is NULL\n updates[\"closed_at\"] = nil\n }\n}\n```","status":"closed","priority":1,"issue_type":"task","created_at":"2025-10-15T14:21:11.837762-07:00","updated_at":"2025-10-15T16:27:28.022212-07:00","closed_at":"2025-10-15T16:27:28.022212-07:00","dependencies":[{"issue_id":"bd-239","depends_on_id":"bd-224","type":"parent-child","created_at":"2025-10-15T14:21:11.838751-07:00","created_by":"stevey"},{"issue_id":"bd-239","depends_on_id":"bd-240","type":"blocks","created_at":"2025-10-15T14:22:36.49138-07:00","created_by":"stevey"}]} {"id":"bd-24","title":"Support ID space partitioning for parallel worker agents","description":"Enable external orchestrators (like AI worker swarms) to control issue ID assignment. Add --id flag to 'bd create' for explicit ID specification. Optionally support 'bd config set next_id N' to set the starting point for auto-increment. Storage layer already supports pre-assigned IDs (sqlite.go:52-71), just need CLI wiring. This keeps beads simple while letting orchestrators implement their own ID partitioning strategies to minimize merge conflicts. Complementary to bd-9's collision resolution.","status":"closed","priority":1,"issue_type":"feature","created_at":"2025-10-14T14:43:06.910467-07:00","updated_at":"2025-10-15T16:27:21.997209-07:00","closed_at":"2025-10-15T03:01:29.569795-07:00"} -{"id":"bd-240","title":"Add CreateIssues interface method to Storage","description":"Add CreateIssues to the Storage interface in storage/storage.go\n\nNon-breaking addition to interface for batch issue creation.","design":"```go\n// storage/storage.go\ntype Storage interface {\n CreateIssue(ctx context.Context, issue *types.Issue, actor string) error\n CreateIssues(ctx context.Context, issues []*types.Issue, actor string) error // NEW\n // ... rest unchanged\n}\n```","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:21.252413-07:00","updated_at":"2025-10-15T16:27:21.997666-07:00","dependencies":[{"issue_id":"bd-240","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:21.253617-07:00","created_by":"stevey"},{"issue_id":"bd-240","depends_on_id":"bd-224","type":"blocks","created_at":"2025-10-15T14:21:21.254504-07:00","created_by":"stevey"}]} +{"id":"bd-240","title":"Add CreateIssues interface method to Storage","description":"Add CreateIssues to the Storage interface in storage/storage.go\n\nNon-breaking addition to interface for batch issue creation.","design":"```go\n// storage/storage.go\ntype Storage interface {\n CreateIssue(ctx context.Context, issue *types.Issue, actor string) error\n CreateIssues(ctx context.Context, issues []*types.Issue, actor string) error // NEW\n // ... rest unchanged\n}\n```","status":"closed","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:21.252413-07:00","updated_at":"2025-10-15T18:30:09.264339-07:00","closed_at":"2025-10-15T18:30:09.264339-07:00","dependencies":[{"issue_id":"bd-240","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:21.253617-07:00","created_by":"stevey"},{"issue_id":"bd-240","depends_on_id":"bd-224","type":"blocks","created_at":"2025-10-15T14:21:21.254504-07:00","created_by":"stevey"}]} {"id":"bd-241","title":"Add comprehensive unit tests for CreateIssues","description":"Test coverage for CreateIssues:\n- Empty batch\n- Single issue\n- Multiple issues\n- Mixed ID assignment (explicit + auto-generated)\n- Validation errors\n- Duplicate ID errors\n- Rollback on error\n- Verify closed_at invariant enforced","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:47.237196-07:00","updated_at":"2025-10-15T16:27:21.998143-07:00","dependencies":[{"issue_id":"bd-241","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:47.246448-07:00","created_by":"stevey"},{"issue_id":"bd-241","depends_on_id":"bd-240","type":"blocks","created_at":"2025-10-15T14:21:47.247811-07:00","created_by":"stevey"}]} {"id":"bd-242","title":"Update import.go to use CreateIssues for bulk imports","description":"Modify cmd/bd/import.go to use CreateIssues instead of CreateIssue loop.\n\nAfter bd-224, import already normalizes closed_at, so this is straightforward:\n1. Normalize all issues in batch (closed_at handling)\n2. Call CreateIssues once with full batch\n3. Much simpler than current loop","design":"```go\n// After normalizing all issues\nfor _, issue := range issues {\n if issue.Status == types.StatusClosed {\n if issue.ClosedAt == nil {\n now := time.Now()\n issue.ClosedAt = \u0026now\n }\n } else {\n issue.ClosedAt = nil\n }\n}\n\n// Single batch call (5-15x faster!)\nreturn store.CreateIssues(ctx, issues, \"import\")\n```","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:47.258493-07:00","updated_at":"2025-10-15T16:27:21.998773-07:00","dependencies":[{"issue_id":"bd-242","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:47.259318-07:00","created_by":"stevey"},{"issue_id":"bd-242","depends_on_id":"bd-240","type":"blocks","created_at":"2025-10-15T14:21:47.25982-07:00","created_by":"stevey"}]} {"id":"bd-243","title":"Document CreateIssues API and update EXTENDING.md","description":"Documentation updates:\n- Godoc for CreateIssues with usage guidance\n- Add batch import examples\n- Update EXTENDING.md with batch usage patterns\n- Performance notes in README.md\n- When to use CreateIssue vs CreateIssues","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:47.398473-07:00","updated_at":"2025-10-15T16:27:21.99948-07:00","dependencies":[{"issue_id":"bd-243","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:47.398904-07:00","created_by":"stevey"},{"issue_id":"bd-243","depends_on_id":"bd-240","type":"blocks","created_at":"2025-10-15T14:21:47.399336-07:00","created_by":"stevey"}]} -{"id":"bd-244","title":"Implement SQLiteStorage.CreateIssues with atomic ID range reservation","description":"Core implementation of CreateIssues in internal/storage/sqlite/sqlite.go\n\nKey optimizations:\n- Single connection + transaction\n- Atomic ID range reservation (generate N IDs in one counter update)\n- Prepared statement for bulk inserts\n- All-or-nothing atomicity\n\nExpected 5-10x speedup for N\u003e10 issues.","design":"Implementation phases per ULTRATHINK_BD222.md:\n\n1. **Validation**: Pre-validate all issues (calls Issue.Validate() which enforces closed_at invariant from bd-224)\n2. **Connection \u0026 Transaction**: BEGIN IMMEDIATE (same as CreateIssue)\n3. **Batch ID Generation**: Reserve range [nextID, nextID+N) in single counter update\n4. **Bulk Insert**: Prepared statement loop (defer multi-VALUE INSERT optimization)\n5. **Bulk Events**: Record creation events for all issues\n6. **Bulk Dirty**: Mark all issues dirty for export\n7. **Commit**: All-or-nothing transaction commit\n\nSee ULTRATHINK_BD222.md lines 344-541 for full implementation details.","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:53.433641-07:00","updated_at":"2025-10-15T16:27:22.000079-07:00","dependencies":[{"issue_id":"bd-244","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:53.435109-07:00","created_by":"stevey"},{"issue_id":"bd-244","depends_on_id":"bd-240","type":"blocks","created_at":"2025-10-15T14:21:53.43563-07:00","created_by":"stevey"},{"issue_id":"bd-244","depends_on_id":"bd-241","type":"blocks","created_at":"2025-10-15T14:22:17.181984-07:00","created_by":"stevey"},{"issue_id":"bd-244","depends_on_id":"bd-242","type":"blocks","created_at":"2025-10-15T14:22:17.195635-07:00","created_by":"stevey"}]} +{"id":"bd-244","title":"Implement SQLiteStorage.CreateIssues with atomic ID range reservation","description":"Core implementation of CreateIssues in internal/storage/sqlite/sqlite.go\n\nKey optimizations:\n- Single connection + transaction\n- Atomic ID range reservation (generate N IDs in one counter update)\n- Prepared statement for bulk inserts\n- All-or-nothing atomicity\n\nExpected 5-10x speedup for N\u003e10 issues.","design":"Implementation phases per ULTRATHINK_BD222.md:\n\n1. **Validation**: Pre-validate all issues (calls Issue.Validate() which enforces closed_at invariant from bd-224)\n2. **Connection \u0026 Transaction**: BEGIN IMMEDIATE (same as CreateIssue)\n3. **Batch ID Generation**: Reserve range [nextID, nextID+N) in single counter update\n4. **Bulk Insert**: Prepared statement loop (defer multi-VALUE INSERT optimization)\n5. **Bulk Events**: Record creation events for all issues\n6. **Bulk Dirty**: Mark all issues dirty for export\n7. **Commit**: All-or-nothing transaction commit\n\nSee ULTRATHINK_BD222.md lines 344-541 for full implementation details.","status":"closed","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:53.433641-07:00","updated_at":"2025-10-15T18:31:28.771539-07:00","closed_at":"2025-10-15T18:31:28.771539-07:00","dependencies":[{"issue_id":"bd-244","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:53.435109-07:00","created_by":"stevey"},{"issue_id":"bd-244","depends_on_id":"bd-240","type":"blocks","created_at":"2025-10-15T14:21:53.43563-07:00","created_by":"stevey"},{"issue_id":"bd-244","depends_on_id":"bd-241","type":"blocks","created_at":"2025-10-15T14:22:17.181984-07:00","created_by":"stevey"},{"issue_id":"bd-244","depends_on_id":"bd-242","type":"blocks","created_at":"2025-10-15T14:22:17.195635-07:00","created_by":"stevey"}]} {"id":"bd-245","title":"Add concurrency tests for CreateIssues","description":"Concurrent testing:\n- Multiple goroutines creating batches in parallel\n- Verify no ID collisions\n- Mix CreateIssue and CreateIssues calls\n- Verify all issues created correctly","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:21:58.802643-07:00","updated_at":"2025-10-15T16:27:22.000481-07:00","dependencies":[{"issue_id":"bd-245","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:21:58.803494-07:00","created_by":"stevey"},{"issue_id":"bd-245","depends_on_id":"bd-244","type":"blocks","created_at":"2025-10-15T14:21:58.804094-07:00","created_by":"stevey"}]} {"id":"bd-247","title":"Add performance benchmarks for CreateIssues","description":"Benchmark suite comparing CreateIssue loop vs CreateIssues batch:\n- 10, 100, 1000 issues\n- Expected: 5-10x speedup for N\u003e10\n- Measure connection, transaction, and insert overhead\n\nTarget: 100 issues in \u003c130ms (vs 900ms sequential)","status":"open","priority":2,"issue_type":"task","created_at":"2025-10-15T14:22:03.391873-07:00","updated_at":"2025-10-15T16:27:22.000882-07:00","dependencies":[{"issue_id":"bd-247","depends_on_id":"bd-222","type":"parent-child","created_at":"2025-10-15T14:22:03.392524-07:00","created_by":"stevey"},{"issue_id":"bd-247","depends_on_id":"bd-244","type":"blocks","created_at":"2025-10-15T14:22:03.392961-07:00","created_by":"stevey"}]} {"id":"bd-248","title":"Test reopen command","description":"","status":"closed","priority":2,"issue_type":"task","created_at":"2025-10-15T16:28:44.246154-07:00","updated_at":"2025-10-15T17:05:23.644788-07:00","closed_at":"2025-10-15T17:05:23.644788-07:00"} diff --git a/internal/storage/sqlite/sqlite.go b/internal/storage/sqlite/sqlite.go index af529def..fa6c5f74 100644 --- a/internal/storage/sqlite/sqlite.go +++ b/internal/storage/sqlite/sqlite.go @@ -504,6 +504,179 @@ func (s *SQLiteStorage) CreateIssue(ctx context.Context, issue *types.Issue, act return nil } +// CreateIssues creates multiple issues atomically in a single transaction. +// This provides significant performance improvements over calling CreateIssue in a loop: +// - Single connection acquisition +// - Single transaction +// - Atomic ID range reservation (one counter update for N issues) +// - All-or-nothing atomicity +// +// Expected 5-10x speedup for batches of 10+ issues. +func (s *SQLiteStorage) CreateIssues(ctx context.Context, issues []*types.Issue, actor string) error { + // Handle empty batch + if len(issues) == 0 { + return nil + } + + // Phase 1: Set timestamps and validate all issues first (fail-fast) + now := time.Now() + for i, issue := range issues { + issue.CreatedAt = now + issue.UpdatedAt = now + + if err := issue.Validate(); err != nil { + return fmt.Errorf("validation failed for issue %d: %w", i, err) + } + } + + // Phase 2: Acquire dedicated connection and start transaction + conn, err := s.db.Conn(ctx) + if err != nil { + return fmt.Errorf("failed to acquire connection: %w", err) + } + defer conn.Close() + + // Begin IMMEDIATE transaction to acquire write lock early + if _, err := conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { + return fmt.Errorf("failed to begin immediate transaction: %w", err) + } + + committed := false + defer func() { + if !committed { + conn.ExecContext(context.Background(), "ROLLBACK") + } + }() + + // Phase 3: Batch ID generation + // Count how many issues need IDs + needIDCount := 0 + for _, issue := range issues { + if issue.ID == "" { + needIDCount++ + } + } + + // Generate ID range atomically if needed + if needIDCount > 0 { + // Get prefix from config + var prefix string + err := conn.QueryRowContext(ctx, `SELECT value FROM config WHERE key = ?`, "issue_prefix").Scan(&prefix) + if err == sql.ErrNoRows || prefix == "" { + prefix = "bd" + } else if err != nil { + return fmt.Errorf("failed to get config: %w", err) + } + + // Atomically reserve ID range: [nextID-needIDCount+1, nextID] + // This is the key optimization - one counter update instead of N + var nextID int + err = conn.QueryRowContext(ctx, ` + INSERT INTO issue_counters (prefix, last_id) + SELECT ?, COALESCE(MAX(CAST(substr(id, LENGTH(?) + 2) AS INTEGER)), 0) + ? + FROM issues + WHERE id LIKE ? || '-%' + AND substr(id, LENGTH(?) + 2) GLOB '[0-9]*' + ON CONFLICT(prefix) DO UPDATE SET + last_id = MAX( + last_id, + (SELECT COALESCE(MAX(CAST(substr(id, LENGTH(?) + 2) AS INTEGER)), 0) + FROM issues + WHERE id LIKE ? || '-%' + AND substr(id, LENGTH(?) + 2) GLOB '[0-9]*') + ) + ? + RETURNING last_id + `, prefix, prefix, needIDCount, prefix, prefix, prefix, prefix, prefix, needIDCount).Scan(&nextID) + if err != nil { + return fmt.Errorf("failed to generate ID range: %w", err) + } + + // Assign IDs sequentially from the reserved range + currentID := nextID - needIDCount + 1 + for i := range issues { + if issues[i].ID == "" { + issues[i].ID = fmt.Sprintf("%s-%d", prefix, currentID) + currentID++ + } + } + } + + // Phase 4: Bulk insert issues using prepared statement + stmt, err := conn.PrepareContext(ctx, ` + INSERT INTO issues ( + id, title, description, design, acceptance_criteria, notes, + status, priority, issue_type, assignee, estimated_minutes, + created_at, updated_at, closed_at, external_ref + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `) + if err != nil { + return fmt.Errorf("failed to prepare statement: %w", err) + } + defer stmt.Close() + + for _, issue := range issues { + _, err = stmt.ExecContext(ctx, + issue.ID, issue.Title, issue.Description, issue.Design, + issue.AcceptanceCriteria, issue.Notes, issue.Status, + issue.Priority, issue.IssueType, issue.Assignee, + issue.EstimatedMinutes, issue.CreatedAt, issue.UpdatedAt, + issue.ClosedAt, issue.ExternalRef, + ) + if err != nil { + return fmt.Errorf("failed to insert issue %s: %w", issue.ID, err) + } + } + + // Phase 5: Bulk record creation events + eventStmt, err := conn.PrepareContext(ctx, ` + INSERT INTO events (issue_id, event_type, actor, new_value) + VALUES (?, ?, ?, ?) + `) + if err != nil { + return fmt.Errorf("failed to prepare event statement: %w", err) + } + defer eventStmt.Close() + + for _, issue := range issues { + eventData, err := json.Marshal(issue) + if err != nil { + // Fall back to minimal description if marshaling fails + eventData = []byte(fmt.Sprintf(`{"id":"%s","title":"%s"}`, issue.ID, issue.Title)) + } + + _, err = eventStmt.ExecContext(ctx, issue.ID, types.EventCreated, actor, string(eventData)) + if err != nil { + return fmt.Errorf("failed to record event for %s: %w", issue.ID, err) + } + } + + // Phase 6: Bulk mark dirty for incremental export + dirtyStmt, err := conn.PrepareContext(ctx, ` + INSERT INTO dirty_issues (issue_id, marked_at) + VALUES (?, ?) + ON CONFLICT (issue_id) DO UPDATE SET marked_at = excluded.marked_at + `) + if err != nil { + return fmt.Errorf("failed to prepare dirty statement: %w", err) + } + defer dirtyStmt.Close() + + dirtyTime := time.Now() + for _, issue := range issues { + _, err = dirtyStmt.ExecContext(ctx, issue.ID, dirtyTime) + if err != nil { + return fmt.Errorf("failed to mark dirty %s: %w", issue.ID, err) + } + } + + // Phase 7: Commit transaction + if _, err := conn.ExecContext(ctx, "COMMIT"); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + committed = true + return nil +} + // GetIssue retrieves an issue by ID func (s *SQLiteStorage) GetIssue(ctx context.Context, id string) (*types.Issue, error) { var issue types.Issue diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 54dc504c..e95e5488 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -11,6 +11,7 @@ import ( type Storage interface { // Issues CreateIssue(ctx context.Context, issue *types.Issue, actor string) error + CreateIssues(ctx context.Context, issues []*types.Issue, actor string) error GetIssue(ctx context.Context, id string) (*types.Issue, error) UpdateIssue(ctx context.Context, id string, updates map[string]interface{}, actor string) error CloseIssue(ctx context.Context, id string, reason string, actor string) error