fix(sharding): add rollback state tracking and improve rollback flow

- Record operation_group metadata in state for observability and resumability
- Enhance rollback sequence to respect dependencies and track executed commands
- Log rollback failures and mark rolled_back status to enable safe resume
- Escape operation_group in queries to prevent injection issues
- Improve error handling to avoid breaking core flow during state writes

Author: donhardman

src/Plugin/Sharding/Queue.php

@@ -91,6 +91,16 @@ public function add(string $nodeId, string $query, string $rollbackQuery, ?strin
 			"
 		);
 
+		// Record operation_group metadata in sharding state for observability/resumability
+		try {
+			$state = new State($this->client);
+			if ($operationGroup !== '') {
+				$state->set("operation_group:{$operationGroup}", ['added_id' => $id, 'created_at' => $mt]);
+			}
+		} catch (\Throwable $_) {
+			// ignore state write failures to avoid breaking core flow
+		}
+
 		return $id;
 	}
 
@@ -346,11 +356,14 @@ public function rollbackOperationGroup(string $operationGroup): bool {
 				return true; // Nothing to rollback is considered success
 			}
 
-			return $this->executeRollbackSequence($rollbackCommands);
+			return $this->executeRollbackSequence($rollbackCommands, $operationGroup);
 		} catch (\Throwable $e) {
 			Buddy::debugvv("Rollback failed for group {$operationGroup}: " . $e->getMessage());
 			return false;
 		}
+	}
+			return false;
+}
 	}
 
 	/**
@@ -359,58 +372,116 @@ public function rollbackOperationGroup(string $operationGroup): bool {
 	 * @return array<array{id:int,node:string,rollback_query:string,rollback_wait_for_id:int}>
 	 *         Rollback commands in reverse execution order
 	 */
-	protected function getRollbackCommands(string $operationGroup): array {
-		$table = $this->cluster->getSystemTableName($this->table);
-
-		// Get completed commands with rollback queries in reverse order
-		$query = "
-			SELECT id, node, rollback_query, rollback_wait_for_id
-			FROM {$table}
-			WHERE operation_group = '{$operationGroup}'
-			AND status = 'processed'
-			AND rollback_query != ''
-			ORDER BY id DESC
-		";
-
-		/** @var array{0?:array{data?:array<array{id:int,node:string,rollback_query:string,rollback_wait_for_id:int}>}} */
-		$result = $this->client->sendRequest($query)->getResult();
-
-		return $result[0]['data'] ?? [];
-	}
+protected function getRollbackCommands(string $operationGroup): array {
+	$table = $this->cluster->getSystemTableName($this->table);
+
+	// Get completed commands with rollback queries in reverse order
+	$escapedGroup = addcslashes($operationGroup, "'");
+	$query = "
+            SELECT id, node, rollback_query, rollback_wait_for_id, operation_group
+            FROM {$table}
+            WHERE operation_group = '{$escapedGroup}'
+            AND status != 'rolled_back'
+            AND rollback_query != ''
+            ORDER BY id DESC
+        ";
+
+	/** @var array{0?:array{data?:array<array{id:int,node:string,rollback_query:string,rollback_wait_for_id:int,operation_group:string}>}} */
+	$result = $this->client->sendRequest($query)->getResult();
+
+	return $result[0]['data'] ?? [];
+}
 
 	/**
 	 * Execute rollback commands in sequence
-	 * @param array<array{id:int,node:string,rollback_query:string,rollback_wait_for_id:int}> $rollbackCommands
-	 * @return bool Success status
+	 * @param array<array{id:int,node:string,rollback_query:string,rollback_wait_for_id:int,operation_group:string}> $rollbackCommands
 	 */
-	protected function executeRollbackSequence(array $rollbackCommands): bool {
-		$allSuccess = true;
+protected function executeRollbackSequence(array $rollbackCommands, string $operationGroup = ''): bool {
+
+	// Keep track of executed ids to satisfy rollback_wait_for_id without DB lookup
+	$executed = [];
+	$allSuccess = true;
+
+	foreach ($rollbackCommands as $command) {
+		$id = (int)$command['id'];
+		$waitFor = (int)$command['rollback_wait_for_id'];
+
+		// If there is dependency, wait until that id is rolled back (or we executed it)
+		if ($waitFor > 0 && !in_array($waitFor, $executed, true)) {
+			while (true) {
+				$dep = $this->getById($waitFor);
+				// If dependency row does not exist, proceed — nothing to wait for
+				if (empty($dep)) {
+					Buddy::debugvv("Rollback: dependency {$waitFor} not found for command {$id}, proceeding");
+					break;
+				}
 
-		foreach ($rollbackCommands as $command) {
-			try {
-				// Execute rollback command on the specific node
-				$nodeId = $command['node'];
-				$rollbackQuery = $command['rollback_query'];
-
-				Buddy::debugvv("Executing rollback on {$nodeId}: {$rollbackQuery}");
-
-				// Execute the rollback query
-				$res = $this->client->sendRequest($rollbackQuery);
-				if ($res->hasError()) {
-					$error = $res->getError();
-					Buddy::debugvv("Rollback command failed: {$rollbackQuery} - Error: {$error}");
-					$allSuccess = false;
-					// Continue with other rollback commands even if one fails
-				} else {
-					Buddy::debugvv("Rollback successful: {$rollbackQuery}");
+				if ($dep['status'] === 'rolled_back' || in_array($waitFor, $executed, true)) {
+					break;
 				}
-			} catch (\Throwable $e) {
-				Buddy::debugvv("Rollback command exception: {$command['rollback_query']} - " . $e->getMessage());
-				$allSuccess = false;
-				// Continue with other rollback commands
+
+				// Sleep briefly to avoid busy loop
+				usleep(100_000); // 100ms
 			}
 		}
 
-		return $allSuccess;
+		try {
+			// Execute rollback command on the specific node
+			$nodeId = $command['node'];
+			$rollbackQuery = $command['rollback_query'];
+
+			Buddy::debugvv("Executing rollback on {$nodeId}: {$rollbackQuery}");
+
+			// Execute the rollback query
+			$res = $this->client->sendRequest($rollbackQuery);
+			if ($res->hasError()) {
+				$error = $res->getError();
+				Buddy::debugvv("Rollback command failed: {$rollbackQuery} - Error: {$error}");
+
+				// Record rollback failure in state and stop sequence so it can be resumed later
+				try {
+					$state = new State($this->client);
+					$state->set("rollback_log:{$operationGroup}:{$id}", $error);
+				} catch (\Throwable $_) {
+					// ignore state write failures
+				}
+
+				return false;
+			}
+
+			Buddy::debugvv("Rollback successful: {$rollbackQuery}");
+
+			// Mark this queue row as rolled_back
+			try {
+				$this->updateStatus($id, 'rolled_back', 0, 0);
+			} catch (\Throwable $u) {
+				Buddy::debugvv("Failed to mark rolled_back for {$id}: " . $u->getMessage());
+				try {
+					$state = new State($this->client);
+					$state->set("rollback_log:{$operationGroup}:{$id}", 'mark_failed:' . $u->getMessage());
+				} catch (\Throwable $_) {
+					// ignore
+				}
+
+				return false;
+			}
+
+			$executed[] = $id;
+		} catch (\Throwable $e) {
+			Buddy::debugvv("Rollback command exception: {$command['rollback_query']} - " . $e->getMessage());
+
+			try {
+				$state = new State($this->client);
+				$state->set("rollback_log:{$operationGroup}:{$id}", $e->getMessage());
+			} catch (\Throwable $_) {
+				// ignore
+			}
+
+			// Stop sequence on exception to allow manual/automated resume
+			return false;
+		}
 	}
+
+	return $allSuccess;
+}
 }

src/Plugin/Sharding/Table.php

@@ -241,6 +241,14 @@ public function shard(
 		$operationGroup = "shard_create_{$this->name}_" . uniqid();
 		$result['operation_group'] = $operationGroup;
 
+		// Record operation_group metadata in state for observability and resumability
+		try {
+			$state = new State($this->client);
+			$state->set("operation_group:{$operationGroup}", json_encode(['table' => $this->name, 'created_at' => time()]));
+		} catch (\Throwable $_) {
+			// ignore state write errors
+		}
+		// Begin main operation scope
 		try {
 			/** @var Map<string,Set<int>> */
 			$nodeShardsMap = new Map;