fix(dynamodb): harden transaction ordering and drop dangling plan references

Addresses review feedback on floci-io#572:

* Replace the delimited-string ordering key for transactWriteItems with a
  Comparator over TransactParticipant tuples. User-supplied PK/SK bytes can
  contain any character, so "storageKey + \0 + itemKey" is not an injective
  ordering — tuple comparison avoids the collision risk entirely.
* Release a table's item locks alongside itemsByTable / itemStore in
  deleteTable so the registry tracks the table's lifetime instead of growing
  unboundedly across create/drop cycles.
* Remove Javadoc and inline comment references to PLAN-dynamodb-concurrency.md,
  which was a local working doc and was never committed. Issue floci-io#571 stays as
  the public reference.

Author: stashev

compatibility-tests/sdk-test-java/src/test/java/com/floci/test/DynamoDbConcurrencyTest.java

@@ -43,8 +43,7 @@
  * End-to-end concurrency tests against a running Floci instance.
  *
  * <p>Covers the four scenarios that must match real DynamoDB's per-item
- * linearisability and transaction atomicity — see issue #571 and
- * PLAN-dynamodb-concurrency.md.
+ * linearisability and transaction atomicity — see issue #571.
  *
  * <p>Each scenario uses a shared {@link CountDownLatch} starting gate so
  * client threads dispatch simultaneously, maximising contention at the server.

src/main/java/io/github/hectorvent/floci/services/dynamodb/DynamoDbService.java

@@ -20,6 +20,7 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -40,8 +41,8 @@ public class DynamoDbService {
     // Items stored per table: storageKey -> Map<itemKey, item>
     // itemKey is "pk" or "pk#sk" depending on table schema
     private final ConcurrentHashMap<String, ConcurrentSkipListMap<String, JsonNode>> itemsByTable = new ConcurrentHashMap<>();
-    // Per-item locks: storageKey -> itemKey -> ReentrantLock. See PLAN-dynamodb-concurrency.md.
-    // Locks are created lazily on first access and never removed; transactWriteItems
+    // Per-item locks: storageKey -> itemKey -> ReentrantLock. Locks are created lazily
+    // on first access and cleared with the table (see deleteTable); transactWriteItems
     // relies on ReentrantLock's re-entrancy so the inner put/update/delete calls do
     // not deadlock after the outer transaction already took each participant's lock.
     private final ConcurrentHashMap<String, ConcurrentHashMap<String, ReentrantLock>> itemLocks = new ConcurrentHashMap<>();
@@ -209,6 +210,7 @@ public void deleteTable(String tableName, String region) {
         }
         tableStore.delete(storageKey);
         itemsByTable.remove(storageKey);
+        itemLocks.remove(storageKey);
         if (itemStore != null) {
             itemStore.delete(storageKey);
         }
@@ -670,12 +672,15 @@ public void transactWriteItems(List<JsonNode> transactItems, String region) {
         // order before evaluating conditions or applying writes. Total-ordered acquisition
         // prevents deadlock across concurrent transactions; ReentrantLock lets the inner
         // putItem/updateItem/deleteItem calls re-enter the same lock for free.
-        TreeMap<String, ReentrantLock> toAcquire = new TreeMap<>();
+        //
+        // Ordering uses a tuple comparator — not a delimited string — so user-supplied
+        // bytes in an item's PK/SK value cannot collide two distinct participants
+        // into the same ordering key.
+        TreeMap<TransactParticipant, ReentrantLock> toAcquire = new TreeMap<>(PARTICIPANT_ORDER);
         for (JsonNode transactItem : transactItems) {
             TransactParticipant p = resolveParticipant(transactItem, region);
             if (p == null) continue;
-            String orderingKey = p.storageKey + "\0" + p.itemKey;
-            toAcquire.putIfAbsent(orderingKey, lockFor(p.storageKey, p.itemKey));
+            toAcquire.putIfAbsent(p, lockFor(p.storageKey, p.itemKey));
         }
 
         List<ReentrantLock> acquired = new ArrayList<>(toAcquire.size());
@@ -737,6 +742,10 @@ public void transactWriteItems(List<JsonNode> transactItems, String region) {
 
     private record TransactParticipant(String storageKey, String itemKey) {}
 
+    private static final Comparator<TransactParticipant> PARTICIPANT_ORDER =
+            Comparator.comparing(TransactParticipant::storageKey)
+                    .thenComparing(TransactParticipant::itemKey);
+
     private TransactParticipant resolveParticipant(JsonNode transactItem, String region) {
         JsonNode target;
         boolean isPut = false;

src/test/java/io/github/hectorvent/floci/services/dynamodb/DynamoDbConcurrencyIntegrationTest.java

@@ -38,7 +38,7 @@
  * regression that surfaces intermittently has a real chance of being observed in a
  * single CI run.
  *
- * <p>See issue #571 and PLAN-dynamodb-concurrency.md.
+ * <p>See issue #571.
  */
 class DynamoDbConcurrencyIntegrationTest {