docs(sharding): clarify rollback and update docs

- Explain rollback usage in sharding context
- Refresh related component documentation

Author: donhardman

doc/sharding/01-components.md

@@ -78,22 +78,20 @@ Manages asynchronous command execution across cluster nodes with rollback suppor
 - **Rollback orchestration** (NEW)
 - **Group-based atomic operations** (NEW)
 
-**Enhanced Methods (NEW):**
-- `addWithRollback()`: Add command with rollback support
-- `rollbackOperationGroup()`: Execute rollback for entire group
-- `getRollbackCommands()`: Retrieve rollback commands
-- `executeRollbackSequence()`: Execute rollback in reverse order
-- `hasRollbackSupport()`: Check for rollback column support
-- `migrateForRollbackSupport()`: Upgrade existing tables
+**Core Methods:**
+- `add(string $nodeId, string $query, string $rollbackQuery, ?string $operationGroup = null)`: Add command with mandatory rollback support
+- `rollbackOperationGroup(string $operationGroup)`: Execute rollback for entire operation group
+- `getRollbackCommands(string $operationGroup)`: Retrieve rollback commands for group
+- `executeRollbackSequence(array $commands)`: Execute rollback commands in reverse order
 
 **Synchronization Pattern:**
 ```php
-// Command A with rollback
-$idA = $queue->addWithRollback($node, "CREATE TABLE t1", "DROP TABLE t1", $group);
+// Command A with rollback (rollback is now mandatory)
+$idA = $queue->add($node, "CREATE TABLE t1", "DROP TABLE IF EXISTS t1", $group);
 
 // Command B waits for A and has rollback
 $queue->setWaitForId($idA);
-$idB = $queue->addWithRollback($node, "ALTER CLUSTER c1 ADD t1", "ALTER CLUSTER c1 DROP t1", $group);
+$idB = $queue->add($node, "ALTER CLUSTER c1 ADD t1", "ALTER CLUSTER c1 DROP t1", $group);
 
 // On failure, rollback entire group
 if ($error) {
@@ -133,17 +131,19 @@ $rebalanceKey = "rebalance:{$tableName}";
 
 Manages cluster topology and node communication.
 
-**Key Features:**
-- Node discovery and health monitoring
-- Cluster configuration management
-- Inter-node communication setup
-- Active/inactive node tracking
+**Key Methods:**
+- `create(?Queue $queue = null, ?string $operationGroup = null)`: Create cluster with rollback support
+- `addNodeIds(Queue $queue, ?string $operationGroup = null, string ...$nodeIds)`: Add nodes to cluster
+- `addTables(Queue $queue, ?string $operationGroup = null, string ...$tables)`: Add tables to cluster
+- `removeTables(Queue $queue, ?string $operationGroup = null, string ...$tables)`: Remove tables from cluster
+- `processPendingTables(Queue $queue, ?string $operationGroup = null)`: Process pending table operations
 
 **Core Responsibilities:**
 - Cluster topology management
 - Node health monitoring
 - Communication channel setup
 - Cluster state synchronization
+- Rollback-aware cluster operations
 
 ### 6. Operator Class (`src/Plugin/Sharding/Operator.php`)
 
@@ -249,8 +249,8 @@ class TestableQueue {
         // Allow null for pure mocking scenarios
     }
 
-    public function add(string $nodeId, string $query): int {
-        return $this->queue?->add($nodeId, $query) ?? 0;
+    public function add(string $nodeId, string $query, string $rollbackQuery, ?string $operationGroup = null): int {
+        return $this->queue?->add($nodeId, $query, $rollbackQuery, $operationGroup) ?? 0;
     }
 
     // Other methods delegate similarly...
@@ -266,43 +266,9 @@ class TestableQueue {
 
 This component architecture provides a robust, scalable foundation for distributed table sharding with comprehensive testing coverage and production-ready reliability.
 
-## New Components (Rollback & Recovery System)
-
-### 7. RollbackCommandGenerator Class (`src/Plugin/Sharding/RollbackCommandGenerator.php`)
-
-Generates reverse SQL commands for rollback operations.
-
-**Key Features:**
-- Automatic rollback command generation
-- Pattern matching for SQL commands
-- Safety checks for rollback operations
-- Batch generation support
-
-**Supported Conversions:**
-- `CREATE TABLE` → `DROP TABLE IF EXISTS`
-- `CREATE CLUSTER` → `DELETE CLUSTER`
-- `ALTER CLUSTER ADD` → `ALTER CLUSTER DROP`
-- `ALTER CLUSTER DROP` → `ALTER CLUSTER ADD`
-- `JOIN CLUSTER` → `DELETE CLUSTER`
-
-**Usage Pattern:**
-```php
-// Direct rollback command provision
-$forwardSql = "CREATE TABLE users (id bigint, name string)";
-$rollbackSql = "DROP TABLE IF EXISTS users";
-$queue->add($nodeId, $forwardSql, $rollbackSql, $operationGroup);
-
-// Common rollback patterns:
-"CREATE TABLE users" → "DROP TABLE IF EXISTS users"
-"CREATE CLUSTER c1" → "DELETE CLUSTER c1"
-"ALTER CLUSTER c1 ADD t1" → "ALTER CLUSTER c1 DROP t1"
-
-// All operations require explicit rollback commands
-$queue->add($node, $sql, $rollbackSql, $operationGroup);
-}
-```
+## New Components (Recovery System)
 
-### 8. CleanupManager Class (`src/Plugin/Sharding/CleanupManager.php`)
+### 7. CleanupManager Class (`src/Plugin/Sharding/CleanupManager.php`)
 
 Manages cleanup of orphaned resources and failed operations.
 
@@ -326,7 +292,7 @@ $results = $cleanup->performFullCleanup();
 // Returns: ['resources_cleaned' => 42, 'actions_taken' => [...]]
 ```
 
-### 9. HealthMonitor Class (`src/Plugin/Sharding/HealthMonitor.php`)
+### 8. HealthMonitor Class (`src/Plugin/Sharding/HealthMonitor.php`)
 
 Monitors system health and performs auto-recovery.
 
@@ -373,7 +339,7 @@ if ($health['overall_status'] !== 'healthy') {
 ```
 Table.shard()
     ├── Creates operation_group
-    ├── Queue.addWithRollback() [multiple times]
+    ├── Queue.add() with rollback [multiple times]
     ├── On failure:
     │   └── Queue.rollbackOperationGroup()
     │       ├── getRollbackCommands()

doc/sharding/04-queue-system.md

@@ -66,16 +66,16 @@ $command = [
 The queue system uses `wait_for_id` to ensure proper command ordering:
 
 ```php
-// Command A
-$idA = $queue->add($node, "COMMAND A");
+// Command A with mandatory rollback
+$idA = $queue->add($node, "COMMAND A", "ROLLBACK A", $operationGroup);
 
 // Command B waits for A to complete
 $queue->setWaitForId($idA);
-$idB = $queue->add($node, "COMMAND B");
+$idB = $queue->add($node, "COMMAND B", "ROLLBACK B", $operationGroup);
 
 // Command C waits for B to complete
 $queue->setWaitForId($idB);
-$idC = $queue->add($node, "COMMAND C");
+$idC = $queue->add($node, "COMMAND C", "ROLLBACK C", $operationGroup);
 
 // Reset dependencies for independent commands
 $queue->resetWaitForId();
@@ -85,19 +85,19 @@ $queue->resetWaitForId();
 
 ### Adding Commands with Rollback
 
-The queue system now supports adding commands with automatic rollback:
+All queue commands now require explicit rollback commands:
 
 ```php
-// Add command with automatic rollback generation
-$queue->addWithRollback(
+// Add command with explicit rollback (rollback is mandatory)
+$queue->add(
     $nodeId,
     "CREATE TABLE users_s0 (id bigint)",
-    null,  // Auto-generate rollback
+    "DROP TABLE IF EXISTS users_s0",  // Explicit rollback required
     $operationGroup
 );
 
-// Add command with explicit rollback
-$queue->addWithRollback(
+// Add cluster command with rollback
+$queue->add(
     $nodeId,
     "ALTER CLUSTER c1 ADD users_s0",
     "ALTER CLUSTER c1 DROP users_s0",  // Explicit rollback
@@ -112,10 +112,10 @@ Related commands are grouped for atomic execution:
 ```php
 $operationGroup = "shard_create_users_" . uniqid();
 
-// All commands in the same group
-$queue->addWithRollback($node1, $cmd1, $rollback1, $operationGroup);
-$queue->addWithRollback($node2, $cmd2, $rollback2, $operationGroup);
-$queue->addWithRollback($node3, $cmd3, $rollback3, $operationGroup);
+// All commands in the same group (rollback required for each)
+$queue->add($node1, $cmd1, $rollback1, $operationGroup);
+$queue->add($node2, $cmd2, $rollback2, $operationGroup);
+$queue->add($node3, $cmd3, $rollback3, $operationGroup);
 
 // On failure, rollback entire group
 if ($error) {
@@ -177,39 +177,41 @@ protected function moveShardWithIntermediateCluster(
     $tempClusterName = "temp_move_{$shardId}_" . uniqid();
 
     // Step 1: Create shard table on target node
-    $createQueueId = $queue->add($targetNode, $this->getCreateTableShardSQL($shardId));
+    $createQueueId = $queue->add($targetNode, $this->getCreateTableShardSQL($shardId), "DROP TABLE IF EXISTS {$shardName}");
 
     // Step 2: Create temporary cluster on SOURCE node (where the data IS)
     // CRITICAL: Use cluster name as path to ensure uniqueness
     $clusterQueueId = $queue->add(
         $sourceNode,
-        "CREATE CLUSTER {$tempClusterName} '{$tempClusterName}' as path"
+        "CREATE CLUSTER {$tempClusterName} '{$tempClusterName}' as path",
+        "DELETE CLUSTER {$tempClusterName}"
     );
 
     // Step 3: Add shard to cluster on SOURCE node FIRST (before JOIN)
     $queue->setWaitForId($clusterQueueId);
-    $queue->add($sourceNode, "ALTER CLUSTER {$tempClusterName} ADD {$shardName}");
+    $queue->add($sourceNode, "ALTER CLUSTER {$tempClusterName} ADD {$shardName}", "ALTER CLUSTER {$tempClusterName} DROP {$shardName}");
 
     // Step 4: TARGET node joins the cluster that SOURCE created
     // Wait for table creation on target node to complete first
     $queue->setWaitForId($createQueueId);
     $joinQueueId = $queue->add(
         $targetNode,
-        "JOIN CLUSTER {$tempClusterName} AT '{$sourceNode}' '{$tempClusterName}' as path"
+        "JOIN CLUSTER {$tempClusterName} AT '{$sourceNode}' '{$tempClusterName}' as path",
+        "DELETE CLUSTER {$tempClusterName}"
     );
 
     // Step 5: CRITICAL - Wait for JOIN to complete (data is now synced)
     // JOIN CLUSTER is synchronous, so once it's processed, data is fully copied
     $queue->setWaitForId($joinQueueId);
-    $dropQueueId = $queue->add($sourceNode, "ALTER CLUSTER {$tempClusterName} DROP {$shardName}");
+    $dropQueueId = $queue->add($sourceNode, "ALTER CLUSTER {$tempClusterName} DROP {$shardName}", "ALTER CLUSTER {$tempClusterName} ADD {$shardName}");
 
     // Step 6: Only after DROP from cluster, remove the table from source
     $queue->setWaitForId($dropQueueId);
-    $deleteQueueId = $queue->add($sourceNode, "DROP TABLE {$shardName}");
+    $deleteQueueId = $queue->add($sourceNode, "DROP TABLE {$shardName}", "");
 
     // Step 7: Clean up temporary cluster ONLY on SOURCE node after all operations complete
     $queue->setWaitForId($deleteQueueId);
-    return $queue->add($sourceNode, "DELETE CLUSTER {$tempClusterName}");
+    return $queue->add($sourceNode, "DELETE CLUSTER {$tempClusterName}", "");
 }
 ```
 
@@ -253,15 +255,15 @@ protected function handleRFNNewNodes(Queue $queue, Vector $schema, Vector $newSc
 
         foreach ($shardsForNewNode as $shard) {
             // Create shard table on new node
-            $queue->add($newNode, $this->getCreateTableShardSQL($shard));
+            $queue->add($newNode, $this->getCreateTableShardSQL($shard), "DROP TABLE IF EXISTS {$shardName}");
 
             // Set up replication (no wait needed - parallel creation)
             $existingNodes = $this->getExistingNodesForShard($schema, $shard);
             $connectedNodes = $existingNodes->merge(new Set([$newNode]));
             $primaryNode = $existingNodes->first();
 
             // Use existing cluster replication mechanism
-            $this->handleReplication($primaryNode, $queue, $connectedNodes, $clusterMap, $shard);
+            $this->handleReplication($primaryNode, $queue, $connectedNodes, $clusterMap, $shard, $operationGroup);
         }
     }
 }
@@ -279,7 +281,7 @@ protected function createDistributedTablesFromSchema(Queue $queue, Vector $newSc
     // Phase 1: Drop all distributed tables with proper force option
     foreach ($newSchema as $row) {
         $sql = "DROP TABLE IF EXISTS {$this->name} OPTION force=1";
-        $queueId = $queue->add($row['node'], $sql);
+        $queueId = $queue->add($row['node'], $sql, "");
         $dropQueueIds->add($queueId);
     }
 
@@ -297,7 +299,7 @@ protected function createDistributedTablesFromSchema(Queue $queue, Vector $newSc
         }
 
         $sql = $this->getCreateShardedTableSQLWithSchema($row['shards'], $newSchema);
-        $queue->add($row['node'], $sql);
+        $queue->add($row['node'], $sql, "DROP TABLE IF EXISTS {$this->name}");
     }
 
     // Reset wait dependencies
@@ -310,14 +312,16 @@ protected function createDistributedTablesFromSchema(Queue $queue, Vector $newSc
 ### Queue Command Addition
 
 ```php
-public function add(string $nodeId, string $query): int {
+public function add(string $nodeId, string $query, string $rollbackQuery, ?string $operationGroup = null): int {
     $queueId = $this->generateNextId();
 
     $command = [
         'id' => $queueId,
         'node' => $nodeId,
         'query' => $query,
+        'rollback_query' => $rollbackQuery,
         'wait_for_id' => $this->currentWaitForId,
+        'operation_group' => $operationGroup ?? '',
         'created_at' => time(),
         'status' => 'pending',
     ];
@@ -373,14 +377,14 @@ public function process(Node $node): void {
 When operations must be strictly ordered:
 
 ```php
-// Pattern: Each operation waits for the previous to complete
-$step1Id = $queue->add($node, "STEP 1 COMMAND");
+// Pattern: Each operation waits for the previous to complete (rollback required)
+$step1Id = $queue->add($node, "STEP 1 COMMAND", "STEP 1 ROLLBACK");
 
 $queue->setWaitForId($step1Id);
-$step2Id = $queue->add($node, "STEP 2 COMMAND");
+$step2Id = $queue->add($node, "STEP 2 COMMAND", "STEP 2 ROLLBACK");
 
 $queue->setWaitForId($step2Id);
-$step3Id = $queue->add($node, "STEP 3 COMMAND");
+$step3Id = $queue->add($node, "STEP 3 COMMAND", "STEP 3 ROLLBACK");
 
 $queue->resetWaitForId();
 ```
@@ -390,35 +394,35 @@ $queue->resetWaitForId();
 When some operations can run in parallel but must converge:
 
 ```php
-// Phase 1: Parallel operations
+// Phase 1: Parallel operations (rollback required for each)
 $queue->resetWaitForId(); // Ensure no dependencies
-$parallel1Id = $queue->add($node1, "PARALLEL COMMAND 1");
-$parallel2Id = $queue->add($node2, "PARALLEL COMMAND 2");
-$parallel3Id = $queue->add($node3, "PARALLEL COMMAND 3");
+$parallel1Id = $queue->add($node1, "PARALLEL COMMAND 1", "PARALLEL ROLLBACK 1");
+$parallel2Id = $queue->add($node2, "PARALLEL COMMAND 2", "PARALLEL ROLLBACK 2");
+$parallel3Id = $queue->add($node3, "PARALLEL COMMAND 3", "PARALLEL ROLLBACK 3");
 
 // Phase 2: Wait for all parallel operations to complete
 $maxParallelId = max($parallel1Id, $parallel2Id, $parallel3Id);
 $queue->setWaitForId($maxParallelId);
 
 // Phase 3: Sequential operations that depend on all parallel operations
-$finalStepId = $queue->add($node1, "FINAL COMMAND");
+$finalStepId = $queue->add($node1, "FINAL COMMAND", "FINAL ROLLBACK");
 ```
 
 ### Cross-Node Synchronization
 
 When operations on different nodes must be coordinated:
 
 ```php
-// Node A prepares
-$prepareId = $queue->add($nodeA, "PREPARE OPERATION");
+// Node A prepares (rollback required)
+$prepareId = $queue->add($nodeA, "PREPARE OPERATION", "PREPARE ROLLBACK");
 
 // Node B waits for Node A to prepare, then joins
 $queue->setWaitForId($prepareId);
-$joinId = $queue->add($nodeB, "JOIN OPERATION");
+$joinId = $queue->add($nodeB, "JOIN OPERATION", "JOIN ROLLBACK");
 
 // Node A waits for Node B to join, then finalizes
 $queue->setWaitForId($joinId);
-$finalizeId = $queue->add($nodeA, "FINALIZE OPERATION");
+$finalizeId = $queue->add($nodeA, "FINALIZE OPERATION", "FINALIZE ROLLBACK");
 ```
 
 ## Error Handling in Queue Operations

doc/sharding/11-rollback-system.md

@@ -56,7 +56,7 @@ CREATE TABLE system.sharding_queue (
 ```
 Table Operations
     ├── Create operation_group
-    ├── Queue.addWithRollback() [multiple commands]
+    ├── Queue.add() with rollback [multiple commands]
     ├── On Success: Mark complete
     └── On Failure:
         └── Queue.rollbackOperationGroup()
@@ -177,7 +177,8 @@ The queue table is automatically created with rollback support:
 
 ```php
 $queue = new Queue($cluster, $client);
-$queue->migrateForRollbackSupport();
+// Queue table now always includes rollback columns
+// No migration needed - rollback is always enabled
 ```
 
 ### Monitoring Setup
@@ -220,9 +221,9 @@ Configure alerts for:
 
 ### Common Issues
 
-1. **Rollback Commands Not Generated**
-   - Check if command is supported by RollbackCommandGenerator
-   - Manually specify rollback command if needed
+1. **Missing Rollback Commands**
+   - All operations now require explicit rollback commands
+   - Provide appropriate rollback SQL when calling Queue::add()
 
 2. **Rollback Execution Fails**
    - Check rollback command syntax