IterateLookup behaves incorrectly after Upsert→Delete→Upsert in v2.0beta3

[KazuhiroNomura]

Describe the bug

Summary

In v2.0beta3, calling IterateLookup after a Upsert → Delete → Upsert sequence causes either a freeze or missing records, depending on the device type. The same logic works correctly in v1.1.

Steps to Reproduce

1. Upsert keys 0–252 (253 records total)
2. Delete all keys 0–252
3. Upsert keys 0–252 again
4. Call IterateLookup in a loop with resetCursor:false; in the Reader callback, return CursorRecordResult.Accept | CursorRecordResult.EndBatch when the batch count reaches 64 (MaxCount)

Expected Behavior (v1.1)

All 253 records (keys 0–252) are returned. The loop exits naturally with TotalCount == 253.

Actual Behavior (v2.0beta3)

Two distinct failure modes are observed depending on the device type.

① LocalMemoryDevice: freeze / infinite loop

IterateLookup never completes. The scan log shows the following repeating pattern, with OnStop firing repeatedly and OnStart never called between batches:

OnStop → Accept... → OnStop
Accept...
OnStop → Accept... → OnStop
OnStop → Accept... → OnStop
(repeats indefinitely)

② Other devices (e.g. ManagedLocalStorageDevice): missing records

The loop terminates, but keys 0–66 (67 records) are never returned. Only keys 67–252 (186 records) are retrieved. Assert.Equal(253, TotalCount) fails.

Comparison with v1.1

  | v1.1 | v2.0beta3 -- | -- | -- Interface | IScanIteratorFunctions | IScanIteratorFunctions Allocator | SpanByteAllocator | ObjectAllocator LocalMemoryDevice | Works correctly | Freezes Other devices | Works correctly | Keys 0–66 missing

Root Cause Hypothesis

The bug appears to be in how v2.0beta3's IterateLookup handles cursor resumption across batch boundaries (EndBatch) when deleted (tombstoned) records are physically present in the log. Specifically, the resume address calculation may not correctly account for skipped delete records, causing the scan to either loop back to an already-visited region (LocalMemoryDevice) or skip the initial segment of live records entirely (other devices). The absence of OnStart between batches in the LocalMemoryDevice log strongly suggests the cursor state is not being preserved correctly when re-entering the scan after EndBatch.

A minimal reproduction is provided in the attached test code.

[KazuhiroNomura]

//Garnet1.1 test project
using System.Diagnostics;
using System.Runtime.InteropServices;
using MemoryPack;
using Tsavorite.core;
using Xunit;
namespace Seamless.Collections.Tests1_1;
using SpanByteStoreFunctions=StoreFunctions<SpanByte,SpanByte,SpanByteComparer,SpanByteRecordDisposer>;
public class Tsavorite1_1{
public class ScanIteratorFunctions<TKey,TValue>:IScanIteratorFunctions<SpanByte,SpanByte> where TKey:struct where TValue:struct{
public const int MaxCount=64;
public readonly(TKey Key,TValue Value)[] Chunk=new (TKey Key,TValue Value)[64];
public readonly List Log=new();
public readonly List<(TKey Key,TValue Value)> Error=new();
public int Count;
public bool SingleReader(ref SpanByte key,ref SpanByte value,RecordMetadata recordMetadata,long numberOfRecords,out CursorRecordResult cursorRecordResult){
var Count=this.Count;
ref readonly var Key = ref MemoryMarshal.AsRef(key.AsReadOnlySpan());
ref readonly var Value=ref MemoryMarshal.AsRef(value.AsReadOnlySpan());
if(Count<this.Chunk.Length){
this.Chunk[Count]=(Key,Value);
} else{
this.Error.Add((Key,Value));
}
Count++;
this.Count=Count;
if(Count==MaxCount){
cursorRecordResult=CursorRecordResult.Accept|CursorRecordResult.EndBatch;
this.Log.Add($"EndBatch {Key},{Value}");
} else{
cursorRecordResult=CursorRecordResult.Accept;
this.Log.Add($"Accept {Key},{Value}");
}
return true;
}
public bool ConcurrentReader(ref SpanByte key,ref SpanByte value,RecordMetadata recordMetadata,long numberOfRecords,out CursorRecordResult cursorRecordResult)
=>this.SingleReader(ref key,ref value,recordMetadata,numberOfRecords,out cursorRecordResult);
public bool OnStart(long beginAddress,long endAddress){
this.Count=0;
this.Log.Add("OnStart");
return true;
}
public void OnException(Exception exception,long numberOfRecords){}
public void OnStop(bool completed,long numberOfRecords){
this.Log.Add("OnStop");
}
}
private const int Count=253;
private const long PageSize=1L<<12;
private const long SegmentSize=PageSize;
private static unsafe void Common0(IDevice LogDevice){
var Settings=new KVSettings<SpanByte,SpanByte>{
IndexSize=1L<<6,
MemorySize=1L<<13,
PageSize=PageSize,
SegmentSize=SegmentSize,
MutableFraction=1,
LogDevice=LogDevice,
};
var StoreFunctions=StoreFunctions<SpanByte,SpanByte>.Create();
var Store=new TsavoriteKV<SpanByte,SpanByte,SpanByteStoreFunctions,SpanByteAllocator>(
Settings,
StoreFunctions,
static (AllocSettings,StoreFuncs)=>new SpanByteAllocator(AllocSettings,StoreFuncs));
using var ReadAddSession=Store.NewSession<SpanByte,SpanByteAndMemory,Empty,SpanByteFunctions>(new());
for(var Key=0;Key<Count;Key++){
var KeyBytes=MemoryPackSerializer.Serialize(Key);
var ValueBytes=MemoryPackSerializer.Serialize(Key);
fixed(byte* kp=KeyBytes,vp=ValueBytes){
var KeySpan=SpanByte.FromPinnedPointer(kp,KeyBytes.Length);
var ValueSpan=SpanByte.FromPinnedPointer(vp,ValueBytes.Length);
ReadAddSession.BasicContext.Upsert(ref KeySpan,ref ValueSpan);
}
}
CompletePendingWithOutputs(ReadAddSession);
for(var Key=0;Key<Count;Key++){
var KeyBytes=MemoryPackSerializer.Serialize(Key);
fixed(byte* kp=KeyBytes){
var KeySpan=SpanByte.FromPinnedPointer(kp,KeyBytes.Length);
ReadAddSession.BasicContext.Delete(ref KeySpan);
}
}
CompletePendingWithOutputs(ReadAddSession);
for(var Key=0;Key<Count;Key++){
var KeyBytes=MemoryPackSerializer.Serialize(Key);
var ValueBytes=MemoryPackSerializer.Serialize(Key);
fixed(byte* kp=KeyBytes,vp=ValueBytes){
var KeySpan=SpanByte.FromPinnedPointer(kp,KeyBytes.Length);
var ValueSpan=SpanByte.FromPinnedPointer(vp,ValueBytes.Length);
ReadAddSession.BasicContext.Upsert(ref KeySpan,ref ValueSpan);
}
}
CompletePendingWithOutputs(ReadAddSession);
var Keys=new List();
var ScanIteratorFunctions=new ScanIteratorFunctions<int,int>();
long TotalCount=0;
long Cursor=0;
while(true){
ScanIteratorFunctions.Count=0;
var r=ReadAddSession.IterateLookup(ref ScanIteratorFunctions,ref Cursor,resetCursor:false);
ScanIteratorFunctions.Log.Add("while");
TotalCount+=ScanIteratorFunctions.Count;
Assert.True(ScanIteratorFunctions.Count<=ScanIteratorFunctions<int,int>.MaxCount);
if(ScanIteratorFunctions.Count<ScanIteratorFunctions<int,int>.MaxCount) break;
}
Keys.Sort();
Assert.Equal(Count,TotalCount);
foreach(var a in ScanIteratorFunctions.Log){
Debug.WriteLine(a);
}
Assert.Empty(ScanIteratorFunctions.Error);
return;
static void CompletePendingWithOutputs(ClientSession<SpanByte,SpanByte,SpanByte,SpanByteAndMemory,Empty,SpanByteFunctions,SpanByteStoreFunctions,SpanByteAllocator> ReadAddSession){
ReadAddSession.BasicContext.CompletePendingWithOutputs(out var outputs,wait:true);
using var completedOutputs=outputs;
while(completedOutputs.Next()){
var status=completedOutputs.Current.Status;
if(status.IsFaulted){
}
}
}
}
[Fact,Trait("Category","Bug")]public void Bug0(){
using var LogDevice=new LocalMemoryDevice(capacity: SegmentSize,sz_segment:SegmentSize,parallelism:1,sector_size:64U,latencyMs:0,fileName:@$"{Environment.CurrentDirectory}\LogDevice.bin");
Common0(LogDevice);
}
[Fact,Trait("Category","Bug")]public unsafe void Bug1(){
using var LogDevice=new ManagedLocalStorageDevice(filename:@$"{Environment.CurrentDirectory}\LogDevice.bin",capacity: SegmentSize,deleteOnClose:true);
Common0(LogDevice);
}
}

[KazuhiroNomura]

//Garnet2.0beta3 test project
using Garnet.common;
using System.Diagnostics;
using System.Runtime.InteropServices;
using MemoryPack;
using Tsavorite.core;
using Xunit;
namespace Seamless.Collections.Tests;

public class Tsavorite2_0{
public class ScanIteratorFunctions<TKey,TValue>:IScanIteratorFunctions where TKey:struct where TValue:struct{
public const int MaxCount=64;
public readonly(TKey Key,TValue Value)[] Chunk=new (TKey Key,TValue Value)[64];
public readonly List Log=new();
public readonly List<(TKey Key,TValue Value)> Error=new();
public int Count;
public bool Reader(in TSourceLogRecord sourceLogRecord,RecordMetadata recordMetadata,long numberOfRecords,out CursorRecordResult cursorRecordResult) where TSourceLogRecord:ISourceLogRecord{
var Count=this.Count;
ref readonly var Key = ref MemoryMarshal.AsRef(sourceLogRecord.Key);
ref readonly var Value=ref MemoryMarshal.AsRef(sourceLogRecord.ValueSpan);
if(Count<this.Chunk.Length){
this.Chunk[Count]=(Key,Value);
} else{
this.Error.Add((Key,Value));
}
Count++;
this.Count=Count;
if(Count==MaxCount){
cursorRecordResult=CursorRecordResult.Accept|CursorRecordResult.EndBatch;
this.Log.Add($"EndBatch {Key},{Value}");
} else{
cursorRecordResult=CursorRecordResult.Accept;
this.Log.Add($"Accept {Key},{Value}");
}
return true;
}

    public bool OnStart(long beginAddress,long endAddress){
        this.Count=0;
        this.Log.Add("OnStart");
        return true;
    }
    public void OnException(Exception exception,long numberOfRecords){}
    public void OnStop(bool completed,long numberOfRecords){
        this.Log.Add("OnStop");
    }
}
private const int Count=253;
private const long PageSize=1L<<12;
private const long SegmentSize=PageSize;
private static void Common0(IDevice LogDevice){
    var Settings=new KVSettings{
        IndexSize=1L<<6,
        LogMemorySize=1L<<13,
        PageSize=PageSize,
        SegmentSize=SegmentSize,
        MutableFraction=1,
        LogDevice=LogDevice,
    };
    var StoreFunctions=new StoreFunctions<SpanByteComparer,DefaultRecordDisposer>(new SpanByteComparer(),()=>null,new DefaultRecordDisposer());
    using var Store=new TsavoriteKV<StoreFunctions<SpanByteComparer,DefaultRecordDisposer>,ObjectAllocator<StoreFunctions<SpanByteComparer,DefaultRecordDisposer>>>(
        Settings,
        StoreFunctions,
        static (AllocSettings,StoreFuncs)=>new ObjectAllocator<StoreFunctions<SpanByteComparer,DefaultRecordDisposer>>(AllocSettings,StoreFuncs));
    using var ReadAddSession=Store.NewSession<FixedSpanByteKey,PinnedSpanByte,SpanByteAndMemory,Empty,SpanByteFunctions<Empty>>(new SpanByteFunctions<Empty>(System.Buffers.MemoryPool<byte>.Shared));
    for(var Key=0;Key<Count;Key++){
        var KeyBytes=MemoryPackSerializer.Serialize(Key);
        var ValueBytes=MemoryPackSerializer.Serialize(Key);
        var Status=ReadAddSession.BasicContext.Upsert((FixedSpanByteKey)KeyBytes,ValueBytes,default);
    }




    CompletePendingWithOutputs(ReadAddSession);
    for(var Key=0;Key<Count;Key++){
        var KeyBytes=MemoryPackSerializer.Serialize(Key);
        var Status=ReadAddSession.BasicContext.Delete((FixedSpanByteKey)KeyBytes,default);
    }



    CompletePendingWithOutputs(ReadAddSession);
    for(var Key=0;Key<Count;Key++){
        var KeyBytes=MemoryPackSerializer.Serialize(Key);
        var ValueBytes=MemoryPackSerializer.Serialize(Key);
        var Status=ReadAddSession.BasicContext.Upsert((FixedSpanByteKey)KeyBytes,ValueBytes,default);
    }




    CompletePendingWithOutputs(ReadAddSession);
    var ScanIteratorFunctions=new ScanIteratorFunctions<int,int>();
    long TotalCount=0;
    long Cursor=0;
    while(true){
        ScanIteratorFunctions.Count=0;
        //Device=LocalMemoryDevice->freeze
        var r=ReadAddSession.IterateLookup(ref ScanIteratorFunctions,ref Cursor,resetCursor:false);
        ScanIteratorFunctions.Log.Add("while");
        TotalCount+=ScanIteratorFunctions.Count;
        if(ScanIteratorFunctions.Count<ScanIteratorFunctions<int,int>.MaxCount) break;
    }
    Assert.Equal(Count,TotalCount);
    //Device=LocalMemoryDevice
    //OnStop->Accept...->OnStop
    //Accept...
    //OnStop->Accept...->OnStop
    //OnStop->Accept...->OnStop
    foreach(var a in ScanIteratorFunctions.Log){
        Debug.WriteLine(a);
    }
    Assert.Empty(ScanIteratorFunctions.Error);
    return;
    static void CompletePendingWithOutputs(ClientSession<FixedSpanByteKey,PinnedSpanByte,SpanByteAndMemory,Empty,SpanByteFunctions<Empty>,StoreFunctions<SpanByteComparer,DefaultRecordDisposer>,ObjectAllocator<StoreFunctions<SpanByteComparer,DefaultRecordDisposer>>> ReadAddSession){
        ReadAddSession.BasicContext.CompletePendingWithOutputs(out var outputs, wait: true);
        using var completedOutputs=outputs;
        while(completedOutputs.Next()){
            var status=completedOutputs.Current.Status;
            if(status.IsFaulted) { }
        }
    }
}
[Fact,Trait("Category","Bug")]public void Bug0(){
    using var LogDevice=new LocalMemoryDevice(capacity:  SegmentSize,sz_segment:SegmentSize,parallelism:1,sector_size:64U,latencyMs:0,fileName:@$"{Environment.CurrentDirectory}\LogDevice.bin");
    Common0(LogDevice);
}
[Fact,Trait("Category","Bug")]public void Bug1(){
    using var LogDevice=new ManagedLocalStorageDevice(
        filename:@$"{Environment.CurrentDirectory}\LogDevice.bin",
        capacity:  SegmentSize,
        deleteOnClose:true
    );
    Common0(LogDevice);
}

}