fix(partition-keys): fence lease release

devel/sql/pgque-api/partition_keys.sql

@@ -446,11 +446,16 @@ create or replace function pgque.release_slot(
 returns boolean as $$
 declare
     v_queue_id int4;
+    v_n int4;
     v_owner text;
+    v_batch_id bigint;
 begin
-    select ps.queue_id, ps.lease_owner into v_queue_id, v_owner
+    select ps.queue_id, ps.lease_owner, pc.n into v_queue_id, v_owner, v_n
     from pgque.partition_slot as ps
     join pgque.queue as q on q.queue_id = ps.queue_id
+    join pgque.partition_consumer as pc
+      on pc.queue_id = ps.queue_id
+     and pc.co_name = ps.co_name
     where q.queue_name = i_queue
       and ps.co_name = i_consumer
       and ps.slot = i_slot
@@ -464,6 +469,15 @@ begin
         return false;
     end if;
 
+    /* The lease is the process-to-ack fence. Clearing it with an open batch
+       would let a successor receive that same batch while this worker may
+       still be processing it. Crash recovery must go through lease expiry. */
+    v_batch_id := pgque._slot_batch(i_queue, i_consumer, i_slot, v_n);
+    if v_batch_id is not null then
+        raise exception 'cannot release slot % of consumer % on queue % while batch % is open; ack the batch first',
+            i_slot, i_consumer, i_queue, v_batch_id;
+    end if;
+
     update pgque.partition_slot
     set lease_owner = null,
         lease_until = null,

devel/sql/pgque-tle.sql

@@ -7633,11 +7633,16 @@ create or replace function pgque.release_slot(
 returns boolean as $$
 declare
     v_queue_id int4;
+    v_n int4;
     v_owner text;
+    v_batch_id bigint;
 begin
-    select ps.queue_id, ps.lease_owner into v_queue_id, v_owner
+    select ps.queue_id, ps.lease_owner, pc.n into v_queue_id, v_owner, v_n
     from pgque.partition_slot as ps
     join pgque.queue as q on q.queue_id = ps.queue_id
+    join pgque.partition_consumer as pc
+      on pc.queue_id = ps.queue_id
+     and pc.co_name = ps.co_name
     where q.queue_name = i_queue
       and ps.co_name = i_consumer
       and ps.slot = i_slot
@@ -7651,6 +7656,15 @@ begin
         return false;
     end if;
 
+    /* The lease is the process-to-ack fence. Clearing it with an open batch
+       would let a successor receive that same batch while this worker may
+       still be processing it. Crash recovery must go through lease expiry. */
+    v_batch_id := pgque._slot_batch(i_queue, i_consumer, i_slot, v_n);
+    if v_batch_id is not null then
+        raise exception 'cannot release slot % of consumer % on queue % while batch % is open; ack the batch first',
+            i_slot, i_consumer, i_queue, v_batch_id;
+    end if;
+
     update pgque.partition_slot
     set lease_owner = null,
         lease_until = null,

devel/sql/pgque.sql

@@ -7543,11 +7543,16 @@ create or replace function pgque.release_slot(
 returns boolean as $$
 declare
     v_queue_id int4;
+    v_n int4;
     v_owner text;
+    v_batch_id bigint;
 begin
-    select ps.queue_id, ps.lease_owner into v_queue_id, v_owner
+    select ps.queue_id, ps.lease_owner, pc.n into v_queue_id, v_owner, v_n
     from pgque.partition_slot as ps
     join pgque.queue as q on q.queue_id = ps.queue_id
+    join pgque.partition_consumer as pc
+      on pc.queue_id = ps.queue_id
+     and pc.co_name = ps.co_name
     where q.queue_name = i_queue
       and ps.co_name = i_consumer
       and ps.slot = i_slot
@@ -7561,6 +7566,15 @@ begin
         return false;
     end if;
 
+    /* The lease is the process-to-ack fence. Clearing it with an open batch
+       would let a successor receive that same batch while this worker may
+       still be processing it. Crash recovery must go through lease expiry. */
+    v_batch_id := pgque._slot_batch(i_queue, i_consumer, i_slot, v_n);
+    if v_batch_id is not null then
+        raise exception 'cannot release slot % of consumer % on queue % while batch % is open; ack the batch first',
+            i_slot, i_consumer, i_queue, v_batch_id;
+    end if;
+
     update pgque.partition_slot
     set lease_owner = null,
         lease_until = null,

tests/test_partition_keys.sql

@@ -430,6 +430,24 @@ begin
   assert cardinality(v_first) = 2,
     format('fencing: zombie must open a 2-event batch, got %s', coalesce(cardinality(v_first), 0));
 
+  /* A cooperative handoff is legal only after ack finishes the batch.
+     Releasing here would let the successor process this same open batch
+     concurrently with the current worker. */
+  v_raised := false;
+  begin
+    perform pgque.release_slot('pk_q', 'w', 0, 'wk-zombie');
+  exception
+    when others then
+      v_raised := true;
+      assert sqlstate = 'P0001'
+        and sqlerrm like 'cannot release slot 0 of consumer w on queue pk_q while batch % is open; ack the batch first',
+        format('fencing: unexpected release error [%s] %s', sqlstate, sqlerrm);
+  end;
+  assert v_raised,
+    'fencing: owner release with an open batch must raise';
+  assert pgque.claim_slot('pk_q', 'w', 0, 'wk-early') is null,
+    'fencing: failed mid-batch release must leave the owner lease intact';
+
   perform pg_sleep(1.2);
 
   -- Heir takes over the expired lease (epoch bump) and is re-issued the