johno/beads
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make merge command idempotent for safe retry after partial failures (bd-26)

Browse Source 
- Added mergeResult struct to track operations (added vs skipped)
- Check if source issues already closed before attempting to close
- Track dependencies migrated vs already existed
- Count text references updated
- Display detailed breakdown of operations in output
- Updated help text to clarify idempotent behavior
- Added comprehensive tests for idempotent retry scenarios
This commit is contained in:
Steve Yegge
2025-10-22 12:24:08 -07:00
parent bafc1a9ded
commit d9e1ee9dd6
3 changed files with 274 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
cmd/bd/merge.go
View File

@@ -18,14 +18,14 @@ var mergeCmd = &cobra.Command{
Short: "Merge duplicate issues into a single issue",
Long: `Merge one or more source issues into a target issue.
This command:
This command is idempotent and safe to retry after partial failures:
1. Validates all issues exist and no self-merge
2. Closes source issues with reason 'Merged into bd-X'
3. Migrates all dependencies from sources to target
4. Updates text references in all issue descriptions/notes
2. Migrates all dependencies from sources to target (skips if already exist)
3. Updates text references in all issue descriptions/notes
4. Closes source issues with reason 'Merged into bd-X' (skips if already closed)
Example:
bd merge bd-42 bd-43 --into bd-42
bd merge bd-42 bd-43 --into bd-41
bd merge bd-10 bd-11 bd-12 --into bd-10 --dry-run`,
Args: cobra.MinimumNArgs(1),
Run: func(cmd *cobra.Command, args []string) {
@@ -62,7 +62,8 @@ Example:
}
// Perform merge
if err := performMerge(ctx, targetID, sourceIDs); err != nil {
result, err := performMerge(ctx, targetID, sourceIDs)
if err != nil {
fmt.Fprintf(os.Stderr, "Error performing merge: %v\n", err)
os.Exit(1)
}
@@ -71,15 +72,23 @@ Example:
markDirtyAndScheduleFlush()
if jsonOutput {
result := map[string]interface{}{
"target_id": targetID,
"source_ids": sourceIDs,
"merged": len(sourceIDs),
output := map[string]interface{}{
"target_id": targetID,
"source_ids": sourceIDs,
"merged": len(sourceIDs),
"dependencies_added": result.depsAdded,
"dependencies_skipped": result.depsSkipped,
"text_references": result.textRefCount,
"issues_closed": result.issuesClosed,
"issues_skipped": result.issuesSkipped,
}
outputJSON(result)
outputJSON(output)
} else {
green := color.New(color.FgGreen).SprintFunc()
fmt.Printf("%s Merged %d issue(s) into %s\n", green("✓"), len(sourceIDs), targetID)
fmt.Printf(" - Dependencies: %d migrated, %d already existed\n", result.depsAdded, result.depsSkipped)
fmt.Printf(" - Text references: %d updated\n", result.textRefCount)
fmt.Printf(" - Source issues: %d closed, %d already closed\n", result.issuesClosed, result.issuesSkipped)
}
},
}
@@ -121,15 +130,26 @@ func validateMerge(targetID string, sourceIDs []string) error {
return nil
}
// mergeResult tracks the results of a merge operation for reporting
type mergeResult struct {
depsAdded int
depsSkipped int
textRefCount int
issuesClosed int
issuesSkipped int
}
// performMerge executes the merge operation
// TODO(bd-202): Add transaction support for atomicity
func performMerge(ctx context.Context, targetID string, sourceIDs []string) error {
func performMerge(ctx context.Context, targetID string, sourceIDs []string) (*mergeResult, error) {
result := &mergeResult{}
// Step 1: Migrate dependencies from source issues to target
for _, sourceID := range sourceIDs {
// Get all dependencies where source is the dependent (source depends on X)
deps, err := store.GetDependencyRecords(ctx, sourceID)
if err != nil {
return fmt.Errorf("failed to get dependencies for %s: %w", sourceID, err)
return nil, fmt.Errorf("failed to get dependencies for %s: %w", sourceID, err)
}
// Migrate each dependency to target
@@ -137,7 +157,7 @@ func performMerge(ctx context.Context, targetID string, sourceIDs []string) erro
// Skip if target already has this dependency
existingDeps, err := store.GetDependencyRecords(ctx, targetID)
if err != nil {
return fmt.Errorf("failed to check target dependencies: %w", err)
return nil, fmt.Errorf("failed to check target dependencies: %w", err)
}
alreadyExists := false
@@ -148,7 +168,9 @@ func performMerge(ctx context.Context, targetID string, sourceIDs []string) erro
}
}
if !alreadyExists && dep.DependsOnID != targetID {
if alreadyExists || dep.DependsOnID == targetID {
result.depsSkipped++
} else {
// Add dependency to target
newDep := &types.Dependency{
IssueID: targetID,
@@ -158,15 +180,16 @@ func performMerge(ctx context.Context, targetID string, sourceIDs []string) erro
CreatedBy: actor,
}
if err := store.AddDependency(ctx, newDep, actor); err != nil {
return fmt.Errorf("failed to migrate dependency %s -> %s: %w", targetID, dep.DependsOnID, err)
return nil, fmt.Errorf("failed to migrate dependency %s -> %s: %w", targetID, dep.DependsOnID, err)
}
result.depsAdded++
}
}
// Get all dependencies where source is the dependency (X depends on source)
allDeps, err := store.GetAllDependencyRecords(ctx)
if err != nil {
return fmt.Errorf("failed to get all dependencies: %w", err)
return nil, fmt.Errorf("failed to get all dependencies: %w", err)
}
for issueID, depList := range allDeps {
@@ -176,7 +199,7 @@ func performMerge(ctx context.Context, targetID string, sourceIDs []string) erro
if err := store.RemoveDependency(ctx, issueID, sourceID, actor); err != nil {
// Ignore "not found" errors as they may have been cleaned up
if !strings.Contains(err.Error(), "not found") {
return fmt.Errorf("failed to remove dependency %s -> %s: %w", issueID, sourceID, err)
return nil, fmt.Errorf("failed to remove dependency %s -> %s: %w", issueID, sourceID, err)
}
}
@@ -192,8 +215,12 @@ func performMerge(ctx context.Context, targetID string, sourceIDs []string) erro
if err := store.AddDependency(ctx, newDep, actor); err != nil {
// Ignore if dependency already exists
if !strings.Contains(err.Error(), "UNIQUE constraint failed") {
return fmt.Errorf("failed to add dependency %s -> %s: %w", issueID, targetID, err)
return nil, fmt.Errorf("failed to add dependency %s -> %s: %w", issueID, targetID, err)
} else {
result.depsSkipped++
}
} else {
result.depsAdded++
}
}
}
@@ -202,27 +229,44 @@ func performMerge(ctx context.Context, targetID string, sourceIDs []string) erro
}
// Step 2: Update text references in all issues
if err := updateMergeTextReferences(ctx, sourceIDs, targetID); err != nil {
return fmt.Errorf("failed to update text references: %w", err)
refCount, err := updateMergeTextReferences(ctx, sourceIDs, targetID)
if err != nil {
return nil, fmt.Errorf("failed to update text references: %w", err)
}
result.textRefCount = refCount
// Step 3: Close source issues
// Step 3: Close source issues (idempotent - skip if already closed)
for _, sourceID := range sourceIDs {
reason := fmt.Sprintf("Merged into %s", targetID)
if err := store.CloseIssue(ctx, sourceID, reason, actor); err != nil {
return fmt.Errorf("failed to close source issue %s: %w", sourceID, err)
issue, err := store.GetIssue(ctx, sourceID)
if err != nil {
return nil, fmt.Errorf("failed to get source issue %s: %w", sourceID, err)
}
if issue == nil {
return nil, fmt.Errorf("source issue not found: %s", sourceID)
}
if issue.Status == types.StatusClosed {
// Already closed - skip
result.issuesSkipped++
} else {
reason := fmt.Sprintf("Merged into %s", targetID)
if err := store.CloseIssue(ctx, sourceID, reason, actor); err != nil {
return nil, fmt.Errorf("failed to close source issue %s: %w", sourceID, err)
}
result.issuesClosed++
}
}
return nil
return result, nil
}
// updateMergeTextReferences updates text references from source IDs to target ID
func updateMergeTextReferences(ctx context.Context, sourceIDs []string, targetID string) error {
// Returns the count of text references updated
func updateMergeTextReferences(ctx context.Context, sourceIDs []string, targetID string) (int, error) {
// Get all issues to scan for references
allIssues, err := store.SearchIssues(ctx, "", types.IssueFilter{})
if err != nil {
return fmt.Errorf("failed to get all issues: %w", err)
return 0, fmt.Errorf("failed to get all issues: %w", err)
}
updatedCount := 0
@@ -284,11 +328,11 @@ func updateMergeTextReferences(ctx context.Context, sourceIDs []string, targetID
// Apply updates if any
if len(updates) > 0 {
if err := store.UpdateIssue(ctx, issue.ID, updates, actor); err != nil {
return fmt.Errorf("failed to update issue %s: %w", issue.ID, err)
return updatedCount, fmt.Errorf("failed to update issue %s: %w", issue.ID, err)
}
updatedCount++
}
}
return nil
return updatedCount, nil
}

199
cmd/bd/merge_test.go
View File

@@ -180,3 +180,202 @@ func containsSubstring(s, substr string) bool {
}
return false
}
// TestPerformMergeIdempotent verifies that merge operations are idempotent
func TestPerformMergeIdempotent(t *testing.T) {
tmpDir := t.TempDir()
dbFile := filepath.Join(tmpDir, ".beads", "issues.db")
if err := os.MkdirAll(filepath.Dir(dbFile), 0755); err != nil {
t.Fatalf("Failed to create test directory: %v", err)
}
testStore, err := sqlite.New(dbFile)
if err != nil {
t.Fatalf("Failed to create test storage: %v", err)
}
defer testStore.Close()
store = testStore
ctx := context.Background()
// Create test issues
issue1 := &types.Issue{
ID: "bd-100",
Title: "Target issue",
Description: "This is the target",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
issue2 := &types.Issue{
ID: "bd-101",
Title: "Source issue 1",
Description: "This mentions bd-100",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
issue3 := &types.Issue{
ID: "bd-102",
Title: "Source issue 2",
Description: "Another source",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
for _, issue := range []*types.Issue{issue1, issue2, issue3} {
if err := testStore.CreateIssue(ctx, issue, "test"); err != nil {
t.Fatalf("Failed to create issue %s: %v", issue.ID, err)
}
}
// Add a dependency from bd-101 to another issue
issue4 := &types.Issue{
ID: "bd-103",
Title: "Dependency target",
Description: "Dependency target",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
if err := testStore.CreateIssue(ctx, issue4, "test"); err != nil {
t.Fatalf("Failed to create issue4: %v", err)
}
dep := &types.Dependency{
IssueID: "bd-101",
DependsOnID: "bd-103",
Type: types.DepBlocks,
}
if err := testStore.AddDependency(ctx, dep, "test"); err != nil {
t.Fatalf("Failed to add dependency: %v", err)
}
// First merge - should complete successfully
result1, err := performMerge(ctx, "bd-100", []string{"bd-101", "bd-102"})
if err != nil {
t.Fatalf("First merge failed: %v", err)
}
if result1.issuesClosed != 2 {
t.Errorf("First merge: expected 2 issues closed, got %d", result1.issuesClosed)
}
if result1.issuesSkipped != 0 {
t.Errorf("First merge: expected 0 issues skipped, got %d", result1.issuesSkipped)
}
if result1.depsAdded == 0 {
t.Errorf("First merge: expected some dependencies added, got 0")
}
// Verify issues are closed
closed1, _ := testStore.GetIssue(ctx, "bd-101")
if closed1.Status != types.StatusClosed {
t.Errorf("bd-101 should be closed after first merge")
}
closed2, _ := testStore.GetIssue(ctx, "bd-102")
if closed2.Status != types.StatusClosed {
t.Errorf("bd-102 should be closed after first merge")
}
// Second merge (retry) - should be idempotent
result2, err := performMerge(ctx, "bd-100", []string{"bd-101", "bd-102"})
if err != nil {
t.Fatalf("Second merge (retry) failed: %v", err)
}
// All operations should be skipped
if result2.issuesClosed != 0 {
t.Errorf("Second merge: expected 0 issues closed, got %d", result2.issuesClosed)
}
if result2.issuesSkipped != 2 {
t.Errorf("Second merge: expected 2 issues skipped, got %d", result2.issuesSkipped)
}
// Dependencies should be skipped (already exist)
if result2.depsAdded != 0 {
t.Errorf("Second merge: expected 0 dependencies added, got %d", result2.depsAdded)
}
// Text references are naturally idempotent - count may vary
// (it will update again but result is the same)
}
// TestPerformMergePartialRetry tests retrying after partial failure
func TestPerformMergePartialRetry(t *testing.T) {
tmpDir := t.TempDir()
dbFile := filepath.Join(tmpDir, ".beads", "issues.db")
if err := os.MkdirAll(filepath.Dir(dbFile), 0755); err != nil {
t.Fatalf("Failed to create test directory: %v", err)
}
testStore, err := sqlite.New(dbFile)
if err != nil {
t.Fatalf("Failed to create test storage: %v", err)
}
defer testStore.Close()
store = testStore
ctx := context.Background()
// Create test issues
issue1 := &types.Issue{
ID: "bd-200",
Title: "Target",
Description: "Target issue",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
issue2 := &types.Issue{
ID: "bd-201",
Title: "Source 1",
Description: "Source 1",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
issue3 := &types.Issue{
ID: "bd-202",
Title: "Source 2",
Description: "Source 2",
Priority: 1,
IssueType: types.TypeTask,
Status: types.StatusOpen,
}
for _, issue := range []*types.Issue{issue1, issue2, issue3} {
if err := testStore.CreateIssue(ctx, issue, "test"); err != nil {
t.Fatalf("Failed to create issue %s: %v", issue.ID, err)
}
}
// Simulate partial failure: manually close one source issue
if err := testStore.CloseIssue(ctx, "bd-201", "Manually closed", "test"); err != nil {
t.Fatalf("Failed to manually close bd-201: %v", err)
}
// Run merge - should handle one already-closed issue gracefully
result, err := performMerge(ctx, "bd-200", []string{"bd-201", "bd-202"})
if err != nil {
t.Fatalf("Merge with partial state failed: %v", err)
}
// Should skip the already-closed issue and close the other
if result.issuesClosed != 1 {
t.Errorf("Expected 1 issue closed, got %d", result.issuesClosed)
}
if result.issuesSkipped != 1 {
t.Errorf("Expected 1 issue skipped, got %d", result.issuesSkipped)
}
// Verify both are now closed
closed1, _ := testStore.GetIssue(ctx, "bd-201")
if closed1.Status != types.StatusClosed {
t.Errorf("bd-201 should remain closed")
}
closed2, _ := testStore.GetIssue(ctx, "bd-202")
if closed2.Status != types.StatusClosed {
t.Errorf("bd-202 should be closed")
}
}