zzstoatzz.io
in-memory redis
redis
in-memory
testing
add XAUTOCLAIM for PEL orphan recovery
Needed by docket's worker to reclaim entries from consumers that died
mid-task — otherwise PEL entries stay claimed forever and a docket
worker restart can't pick up the stuck work.
Mirrors upstream src/store.rs::xautoclaim:
- Walks every consumer's PEL in the named group.
- Filters by min_idle_ms and `start` cursor.
- Sorts by stream ID for deterministic pagination.
- Transfers each qualifying entry to the claiming consumer, even
when the underlying stream entry was trimmed (reports it in
deleted_ids so the caller can XACK and drain the PEL).
- Returns (next_start, claimed entries, deleted_ids); next_start ==
zero means we processed everything in range.
Test covers the basic transfer path (c1 reads, c2 autoclaims with
min_idle=0 → all moved). 41/41 tests pass.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Opus 4.7 (1M context)
date
(May 23, 2026, 7:39 PM -0500)
commit
1d1bba6e
1d1bba6e00787a618c94e0019df28f380d4de517
parent
dcfca159
dcfca1597cbbffd94b1eb7d4a51baaf99f07dbba
+151
+151
src/store.zig
+151
src/store.zig
···
1815
1815
return entry.data.stream.last_id;
1816
1816
}
1817
1817
1818
+
/// Result of an XAUTOCLAIM call. Caller owns the byte buffers via
1819
+
/// freeAutoclaimResult.
1820
+
pub const AutoclaimResult = struct {
1821
+
/// Next start ID for pagination. `StreamId.zero()` means all
1822
+
/// qualifying entries processed.
1823
+
next_start: StreamId,
1824
+
/// Entries that still exist in the stream and have been transferred
1825
+
/// to the claiming consumer.
1826
+
claimed: []DeliveredEntry,
1827
+
/// IDs that were in some consumer's PEL but no longer in the stream
1828
+
/// (likely trimmed). Caller should XACK these.
1829
+
deleted_ids: []StreamId,
1830
+
};
1831
+
1832
+
pub fn freeAutoclaimResult(alloc: Allocator, r: AutoclaimResult) void {
1833
+
for (r.claimed) |de| Store.freeHashPairs(alloc, de.fields);
1834
+
alloc.free(r.claimed);
1835
+
alloc.free(r.deleted_ids);
1836
+
}
1837
+
1838
+
/// XAUTOCLAIM: reclaim idle pending entries from any consumer and
1839
+
/// transfer them to `consumer`. Returns the next pagination cursor,
1840
+
/// the (now reassigned) entries, and IDs whose stream entries no
1841
+
/// longer exist (caller should XACK those to drain the PEL).
1842
+
pub fn xautoclaim(
1843
+
self: *Store,
1844
+
out_alloc: Allocator,
1845
+
key: []const u8,
1846
+
group: []const u8,
1847
+
consumer: []const u8,
1848
+
min_idle_ms: u64,
1849
+
start: StreamId,
1850
+
count: ?usize,
1851
+
) StoreError!AutoclaimResult {
1852
+
self.mutex.lockUncancelable(self.io);
1853
+
defer self.mutex.unlock(self.io);
1854
+
1855
+
const entry = self.liveEntry(key) orelse return StoreError.NoGroup;
1856
+
if (entry.data != .stream) return StoreError.WrongType;
1857
+
const s = &entry.data.stream;
1858
+
const cg = s.groups.getPtr(group) orelse return StoreError.NoGroup;
1859
+
1860
+
const now = self.nowNs();
1861
+
const min_idle_ns: i128 = @as(i128, min_idle_ms) * std.time.ns_per_ms;
1862
+
1863
+
// Collect qualifying (id, dc, original_consumer_owned_key) from
1864
+
// every consumer's PEL.
1865
+
const Qual = struct {
1866
+
id: StreamId,
1867
+
dc: u64,
1868
+
owner: []const u8,
1869
+
};
1870
+
var qual: std.ArrayList(Qual) = .empty;
1871
+
defer qual.deinit(out_alloc);
1872
+
1873
+
var cons_it = cg.consumers.iterator();
1874
+
while (cons_it.next()) |cons_kv| {
1875
+
const owner = cons_kv.key_ptr.*;
1876
+
var pit = cons_kv.value_ptr.pending.iterator();
1877
+
while (pit.next()) |pe| {
1878
+
if (StreamId.cmp(pe.key_ptr.*, start) == .lt) continue;
1879
+
if ((now - pe.value_ptr.delivered_at_ns) < min_idle_ns) continue;
1880
+
try qual.append(out_alloc, .{
1881
+
.id = pe.key_ptr.*,
1882
+
.dc = pe.value_ptr.delivery_count,
1883
+
.owner = owner,
1884
+
});
1885
+
}
1886
+
}
1887
+
std.mem.sort(Qual, qual.items, {}, struct {
1888
+
fn lt(_: void, a: Qual, b: Qual) bool {
1889
+
return StreamId.cmp(a.id, b.id) == .lt;
1890
+
}
1891
+
}.lt);
1892
+
1893
+
const want = count orelse std.math.maxInt(usize);
1894
+
const processed: usize = @min(qual.items.len, want);
1895
+
const next_start: StreamId = if (processed < qual.items.len)
1896
+
qual.items[processed].id
1897
+
else
1898
+
StreamId.zero();
1899
+
1900
+
var claimed: std.ArrayList(DeliveredEntry) = .empty;
1901
+
errdefer {
1902
+
for (claimed.items) |de| Store.freeHashPairs(out_alloc, de.fields);
1903
+
claimed.deinit(out_alloc);
1904
+
}
1905
+
var deleted: std.ArrayList(StreamId) = .empty;
1906
+
errdefer deleted.deinit(out_alloc);
1907
+
1908
+
var idx: usize = 0;
1909
+
while (idx < processed) : (idx += 1) {
1910
+
const q = qual.items[idx];
1911
+
// Remove from original consumer's PEL.
1912
+
if (cg.consumers.getPtr(q.owner)) |orig| {
1913
+
_ = orig.pending.remove(q.id);
1914
+
}
1915
+
1916
+
// If the entry still exists in the stream, add to claimed.
1917
+
if (streamFindEntry(s, q.id)) |se| {
1918
+
try claimed.append(out_alloc, .{
1919
+
.id = q.id,
1920
+
.fields = try dupeStreamFields(out_alloc, se.fields),
1921
+
});
1922
+
} else {
1923
+
try deleted.append(out_alloc, q.id);
1924
+
}
1925
+
1926
+
// Auto-create the claiming consumer.
1927
+
if (!cg.consumers.contains(consumer)) {
1928
+
const ck = try self.allocator.dupe(u8, consumer);
1929
+
errdefer self.allocator.free(ck);
1930
+
try cg.consumers.put(ck, Consumer.init(self.allocator));
1931
+
}
1932
+
const target = cg.consumers.getPtr(consumer).?;
1933
+
try target.pending.put(q.id, .{
1934
+
.delivered_at_ns = now,
1935
+
.delivery_count = q.dc + 1,
1936
+
});
1937
+
}
1938
+
1939
+
return .{
1940
+
.next_start = next_start,
1941
+
.claimed = try claimed.toOwnedSlice(out_alloc),
1942
+
.deleted_ids = try deleted.toOwnedSlice(out_alloc),
1943
+
};
1944
+
}
1945
+
1818
1946
pub fn xgroupDestroy(self: *Store, key: []const u8, group: []const u8) StoreError!bool {
1819
1947
self.mutex.lockUncancelable(self.io);
1820
1948
defer self.mutex.unlock(self.io);
···
2527
2655
}
2528
2656
try std.testing.expectEqual(@as(i64, 3), try s.xtrimMaxlen("st", 2));
2529
2657
try std.testing.expectEqual(@as(i64, 2), try s.xlen("st"));
2658
+
}
2659
+
2660
+
test "XAUTOCLAIM transfers idle PEL entries" {
2661
+
var threaded = Io.Threaded.init(std.testing.allocator, .{});
2662
+
defer threaded.deinit();
2663
+
var s = Store.init(std.testing.allocator, threaded.io());
2664
+
defer s.deinit();
2665
+
2666
+
_ = try s.xadd("st", .{ .ms = 1, .seq = 0 }, &.{ "k", "v1" });
2667
+
_ = try s.xadd("st", .{ .ms = 2, .seq = 0 }, &.{ "k", "v2" });
2668
+
try s.xgroupCreate("st", "g1", .{ .id = StreamId.zero() });
2669
+
2670
+
// c1 reads → claims both into c1's PEL
2671
+
const got = try s.xreadgroupOne(std.testing.allocator, "st", "g1", "c1", .new_messages, 10);
2672
+
defer freeDeliveredEntries(std.testing.allocator, got);
2673
+
try std.testing.expectEqual(@as(usize, 2), got.len);
2674
+
2675
+
// c2 autoclaims with min_idle=0 → reclaims everything
2676
+
const r = try s.xautoclaim(std.testing.allocator, "st", "g1", "c2", 0, StreamId.zero(), null);
2677
+
defer Store.freeAutoclaimResult(std.testing.allocator, r);
2678
+
try std.testing.expectEqual(@as(usize, 2), r.claimed.len);
2679
+
try std.testing.expectEqual(@as(usize, 0), r.deleted_ids.len);
2680
+
try std.testing.expect(StreamId.eql(r.next_start, StreamId.zero()));
2530
2681
}
2531
2682
2532
2683
test "XGROUP DESTROY removes the group" {