Fix/trashcan restore wal barrier

embedded/graph/crud/ll_crud.go

@@ -82,7 +82,11 @@
 		newDownstreamPayload.SetByPath("__parent_holds_locks", parentHoldLocks)
 	}
 
+	// Lock bookkeeping belongs to this handler invocation. The held locks are
+	// promoted above; the mark must not travel, or the child unlock would consume
+	// the parent's WAL-barrier mark while the parent is still writing.
 	newDownstreamPayload.RemoveByPath("__key_locks")
+	newDownstreamPayload.RemoveByPath("__key_lock_time")
 	return &newDownstreamPayload
 }
 
@@ -121,6 +125,37 @@
 	ctx.Payload.SetByPath(fmt.Sprintf("__key_locks.%s.k", seg), easyjson.NewJSON(key))
 }
 
+// operationActiveMarkState returns the mark held by the current lock span.
+func operationActiveMarkState(payload *easyjson.JSON) (opTime int64, held bool) {
+	if payload == nil {
+		return 0, false
+	}
+	return payload.GetByPath("__key_lock_time").AsInt64()
+}
+
+// markOperationActiveOnce balances several lock passes followed by one unlock
+// with a single activeOps mark. If a later pass carries an older opTime, move
+// the mark down without leaving a momentarily empty barrier.
+func markOperationActiveOnce(ctx *sfPlugins.StatefunContextProcessor, opTime int64) {
+	prev, held := operationActiveMarkState(ctx.Payload)
+	switch {
+	case !held:
+		// The common path: first write-lock pass of this invocation.
+	case opTime >= prev:
+		// The older mark already covers this pass.
+		return
+	default:
+		// Unexpected today; activate the older mark before releasing the newer.
+		lg.Logf(lg.WarnLevel, "markOperationActiveOnce: nested lock pass uses opTime=%d, earlier than the held mark %d; migrating the barrier down", opTime, prev)
+		ctx.Domain.Cache().MarkOperationActive(opTime)
+		ctx.Domain.Cache().MarkOperationDone(prev)
+		ctx.Payload.SetByPath("__key_lock_time", easyjson.NewJSON(opTime))
+		return
+	}
+	ctx.Payload.SetByPath("__key_lock_time", easyjson.NewJSON(opTime))
+	ctx.Domain.Cache().MarkOperationActive(opTime)
+}
+
 func parentHoldsWriteLock(ctx *sfPlugins.StatefunContextProcessor, key string) bool {
 	return ctx.Payload.GetByPath(fmt.Sprintf("__parent_holds_locks.%s.m", lockRecSeg(key))).AsStringDefault("") == "w"
 }
@@ -188,8 +223,7 @@
 		}
 	}
 	if lockedWriteAny {
-		ctx.Payload.SetByPath("__key_lock_time", easyjson.NewJSON(opTime))
-		ctx.Domain.Cache().MarkOperationActive(opTime)
+		markOperationActiveOnce(ctx, opTime)
 	}
 }
 
@@ -244,8 +278,7 @@
 		}
 	}
 	if lockedWriteAny {
-		ctx.Payload.SetByPath("__key_lock_time", easyjson.NewJSON(opTime))
-		ctx.Domain.Cache().MarkOperationActive(opTime)
+		markOperationActiveOnce(ctx, opTime)
 	}
 }
 
@@ -266,13 +299,8 @@
 		}
 		ctx.Payload.RemoveByPath("__key_locks")
 	}
-	// Completion marking is deliberately DECOUPLED from the lock records:
-	// __key_lock_time is a flat field that always parses, and it is set exactly
-	// when MarkOperationActive was called — so even a bookkeeping bug that
-	// loses a lock record can orphan at most that one lock, never the
-	// activeOps entry (an orphaned entry wedges the WAL publisher forever).
-	// The field is consumed so repeated lock/unlock pairs within one handler
-	// stay symmetric.
+	// Completion is independent of lock-record parsing. The flat field is
+	// consumed once; child forwarding strips it and repeated lock passes reuse it.
 	if opTime, ok := ctx.Payload.GetByPath("__key_lock_time").AsInt64(); ok {
 		ctx.Domain.Cache().MarkOperationDone(opTime)
 		ctx.Payload.RemoveByPath("__key_lock_time")
@@ -849,8 +877,8 @@
 					}
 				}
 				if brokenTarget {
 					outLinkTypes = append(outLinkTypes)
 					outLinkIds = append(outLinkIds)
 				}
 			}
 			result.SetByPath("links.out.names", easyjson.NewJSON(outLinkNames))

embedded/graph/crud/ll_crud_lock_marking_test.go

@@ -0,0 +1,67 @@
+package crud
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/foliagecp/easyjson"
+	sfPlugins "github.com/foliagecp/sdk/statefun/plugins"
+	"github.com/stretchr/testify/require"
+)
+
+func ctxWithPayload(p easyjson.JSON) *sfPlugins.StatefunContextProcessor {
+	return &sfPlugins.StatefunContextProcessor{Payload: &p}
+}
+
+func Test_InjectParentHoldsLocks_StripsPerInvocationBookkeeping(t *testing.T) {
+	payload := easyjson.NewJSONObject()
+	ctx := ctxWithPayload(payload)
+	recordHeldLock(ctx, "some/vertex", "w")
+	ctx.Payload.SetByPath("__key_lock_time", easyjson.NewJSON(int64(12345)))
+	ctx.Payload.SetByPath("op_time", easyjson.NewJSON(int64(12345)))
+
+	downstream := injectParentHoldsLocks(ctx, ctx.Payload)
+
+	seg := lockRecSeg("some/vertex")
+	require.Equal(t, "w", downstream.GetByPath(fmt.Sprintf("__parent_holds_locks.%s.m", seg)).AsStringDefault(""))
+	require.Equal(t, "some/vertex", downstream.GetByPath(fmt.Sprintf("__parent_holds_locks.%s.k", seg)).AsStringDefault(""))
+	require.False(t, downstream.PathExists("__key_locks"))
+	require.False(t, downstream.PathExists("__key_lock_time"))
+	require.True(t, downstream.PathExists("op_time"))
+}
+
+func Test_InjectParentHoldsLocks_StripsLockTimeFromAClonedPayload(t *testing.T) {
+	payload := easyjson.NewJSONObject()
+	ctx := ctxWithPayload(payload)
+	recordHeldLock(ctx, "owner/vertex", "w")
+	ctx.Payload.SetByPath("__key_lock_time", easyjson.NewJSON(int64(777)))
+
+	cloned := ctx.Payload.Clone()
+	require.True(t, cloned.PathExists("__key_lock_time"))
+
+	downstream := injectParentHoldsLocks(ctx, &cloned)
+
+	require.False(t, downstream.PathExists("__key_lock_time"))
+	require.False(t, downstream.PathExists("__key_locks"))
+}
+
+func Test_OperationActiveMarkState(t *testing.T) {
+	t.Run("absent", func(t *testing.T) {
+		payload := easyjson.NewJSONObject()
+		_, held := operationActiveMarkState(&payload)
+		require.False(t, held)
+	})
+
+	t.Run("present", func(t *testing.T) {
+		payload := easyjson.NewJSONObject()
+		payload.SetByPath("__key_lock_time", easyjson.NewJSON(int64(42)))
+		opTime, held := operationActiveMarkState(&payload)
+		require.True(t, held)
+		require.Equal(t, int64(42), opTime)
+	})
+
+	t.Run("nil payload", func(t *testing.T) {
+		_, held := operationActiveMarkState(nil)
+		require.False(t, held)
+	})
+}

embedded/graph/crud/trash_can.go

@@ -153,10 +153,18 @@ func trashCanEdgeInfo(ctx *sfPlugins.StatefunContextProcessor, selfID string) (o
 // is flagged with a WARNING — an object's identity carries its type, so this
 // usually means the id got reused or the model changed underneath the user.
 //
-// Caller (createObjectInline) holds the object's write lock; the trash links
-// are removed under additional per-edge write locks accumulated into the same
-// lock set (released by the caller's operationKeysMutexUnlock).
+// createObjectInline (from CreateObject or the UpdateObject upsert path) already
+// holds the object's write lock and activeOps mark. Extra edge locks join that
+// lock set and are released by the caller's single operationKeysMutexUnlock;
+// markOperationActiveOnce prevents this second pass from adding another mark.
 func restoreObjectFromTrashCan(ctx *sfPlugins.StatefunContextProcessor, selfID, requestedType string, incomingBody *easyjson.JSON, opTime int64) {
+	// Without the caller's mark, the edge locks below belong to no unlock.
+	if _, held := operationActiveMarkState(ctx.Payload); !held {
+		lg.Logf(lg.WarnLevel,
+			"trash can restore: object %s restored without the caller's write lock and operation mark; the edge locks taken here will be released by nobody",
+			selfID)
+	}
+
 	trashType := trashCanTypeID(ctx)
 	linkName := ctx.Domain.GetObjectIDWithoutDomain(selfID)
 

embedded/graph/crud/trash_can_wal_barrier_test.go

@@ -0,0 +1,60 @@
+package crud_test
+
+import (
+	"time"
+
+	"github.com/foliagecp/easyjson"
+)
+
+// An orphaned activeOps entry prevents pendingTxs from draining.
+func (s *TrashCanTestSuite) requireCacheQuiesced(what string) {
+	s.T().Helper()
+
+	cs := s.Runtime().Domain.Cache()
+	st := cs.StatsForTest()
+
+	deadline := time.Now().Add(20 * time.Second)
+	for time.Now().Before(deadline) {
+		st = cs.StatsForTest()
+		if st.ActiveOps == 0 && st.PendingTxs == 0 {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	s.T().Fatalf("cache did not quiesce after %s: activeOps=%d pendingTxs=%d",
+		what, st.ActiveOps, st.PendingTxs)
+}
+
+func (s *TrashCanTestSuite) Test_Restore_LeavesNoOrphanedActiveOperation() {
+	s.boot()
+	s.NoError(s.cmdb.TypeCreate("tcb_t"))
+	s.requireCacheQuiesced("type create")
+
+	s.NoError(s.cmdb.ObjectCreate("tcb1", "tcb_t", usrBody("h1", "alice", "prod")))
+	s.requireCacheQuiesced("object create")
+
+	s.NoError(s.cmdb.ObjectDelete("tcb1"))
+	s.requireCacheQuiesced("object delete (park)")
+
+	s.NoError(s.cmdb.ObjectCreate("tcb1", "tcb_t", easyjson.NewJSONObjectWithKeyValue("hostname", easyjson.NewJSON("h2"))))
+	s.requireCacheQuiesced("object create (restore from trash can)")
+
+	// An orphan from restore would block this later transaction.
+	s.NoError(s.cmdb.ObjectCreate("tcb1-after", "tcb_t", easyjson.NewJSONObjectWithKeyValue("hostname", easyjson.NewJSON("h3"))))
+	s.requireCacheQuiesced("write after restore")
+}
+
+func (s *TrashCanTestSuite) Test_UpsertRestore_LeavesNoOrphanedActiveOperation() {
+	s.boot()
+	s.NoError(s.cmdb.TypeCreate("tcb_t"))
+	s.NoError(s.cmdb.ObjectCreate("tcb2", "tcb_t", usrBody("h1", "bob")))
+	s.NoError(s.cmdb.ObjectDelete("tcb2"))
+	s.requireCacheQuiesced("object delete (park)")
+
+	s.NoError(s.cmdb.ObjectUpdate("tcb2", easyjson.NewJSONObjectWithKeyValue("hostname", easyjson.NewJSON("h2")), true, "tcb_t"))
+	s.requireCacheQuiesced("object upsert (restore from trash can)")
+
+	s.NoError(s.cmdb.ObjectCreate("tcb2-after", "tcb_t", easyjson.NewJSONObjectWithKeyValue("hostname", easyjson.NewJSON("h3"))))
+	s.requireCacheQuiesced("write after upsert-restore")
+}

scripts/run-leak-tests.sh

@@ -7,7 +7,7 @@
 #   --mode quick     W=2 M=8 cycles, small workloads (default; smoke, ~15-20 min for 'all')
 #   --mode full      W=5 M=20 cycles, 3x workloads, tighter floors — the "3-sigma claim" run
 #   --mode soak      dispatch to scripts/run-soak-tests.sh --scenario leak-hunt (docker)
-#   --scenario NAME  s0..s12, 'core' (s0 s1 s2 s5 s9), or 'all' (default)
+#   --scenario NAME  s0..s14, 'core' (s0 s1 s2 s5 s9), or 'all' (default)
 #   --results DIR    artifacts root (default tests/leak/_results/leak-<UTC>/)
 #   --race           run the Go suite under the race detector
 #
@@ -87,6 +87,7 @@ rx_for() {
     s11) echo '^TestS11ExportSessions$' ;;
     s12) echo '^TestS12KVGrowthReport$' ;;
     s13) echo '^TestS13SaltedHLChurn$' ;;
+    s14) echo '^TestS14TrashCanRestore$' ;;
     core) echo '^(TestS0PlantedLeakIsFlagged|TestS0ControlIsClean|TestS1LLCrudChurn|TestS2CMDBObjectChurn|TestS5JPGQL|TestS9CacheStore)$' ;;
     all) echo '^TestS[0-9]' ;;
     *) return 1 ;;
@@ -99,9 +100,9 @@ rx_for() {
 # process per scenario = clean baselines, zero cross-contamination. The test
 # binary is compiled once and reused by go test's build cache.
 case "$SCENARIO" in
-  all)  SCEN_LIST="s0 s1 s2 s3 s4 s5 s6 s7 s8 s9 s10 s11 s12 s13" ;;
+  all)  SCEN_LIST="s0 s1 s2 s3 s4 s5 s6 s7 s8 s9 s10 s11 s12 s13 s14" ;;
   core) SCEN_LIST="s0 s1 s2 s5 s9" ;;
-  *)    rx_for "$SCENARIO" >/dev/null || { echo "unknown scenario: $SCENARIO (s0..s12|core|all)"; exit 2; }
+  *)    rx_for "$SCENARIO" >/dev/null || { echo "unknown scenario: $SCENARIO (s0..s14|core|all)"; exit 2; }
         SCEN_LIST="$SCENARIO" ;;
 esac