dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add checksum validation on MsgBlock read path (Gap 1.5)

Browse Source 
Add _lastChecksum field and LastChecksum property to MsgBlock tracking
the XxHash64 checksum of the last written record (Go: msgBlock.lchk,
filestore.go:2204). Capture the checksum from the encoded record trailer
on every Write/WriteAt/WriteSkip call. Read-path validation happens
naturally through the existing MessageRecord.Decode checksum check.
This commit is contained in:
Joseph Doherty
2026-02-25 07:50:03 -05:00
parent 9ac29fc6f5
commit 5beeb1b3f6
2 changed files with 131 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/NATS.Server/JetStream/Storage/MsgBlock.cs
View File

@@ -43,6 +43,11 @@ public sealed class MsgBlock : IDisposable
// Reference: golang/nats-server/server/filestore.go:236 (cache field)
private Dictionary<ulong, MessageRecord>? _cache;
// Go: msgBlock.lchk — last written record checksum (XxHash64, 8 bytes).
// Tracked so callers can chain checksum verification across blocks.
// Reference: golang/nats-server/server/filestore.go:2204 (lchk field)
private byte[]? _lastChecksum;
private MsgBlock(FileStream file, int blockId, long maxBytes, ulong firstSequence)
{
_file = file;
@@ -137,6 +142,22 @@ public sealed class MsgBlock : IDisposable
}
}
/// <summary>
/// The XxHash64 checksum of the last record written to this block (8 bytes), or null
/// if no records have been written yet. Updated after every <see cref="Write"/>,
/// <see cref="WriteAt"/>, or <see cref="WriteSkip"/> call.
/// Reference: golang/nats-server/server/filestore.go:2204 (msgBlock.lchk)
/// </summary>
public byte[]? LastChecksum
{
get
{
_lock.EnterReadLock();
try { return _lastChecksum is null ? null : (byte[])_lastChecksum.Clone(); }
finally { _lock.ExitReadLock(); }
}
}
/// <summary>
/// Creates a new empty block file.
/// </summary>
@@ -215,6 +236,11 @@ public sealed class MsgBlock : IDisposable
_cache ??= new Dictionary<ulong, MessageRecord>();
_cache[sequence] = record;
// Go: msgBlock.lchk — capture checksum (last 8 bytes of encoded record).
// Reference: golang/nats-server/server/filestore.go:2204 (lchk update on write)
_lastChecksum ??= new byte[8];
encoded.AsSpan(^8..).CopyTo(_lastChecksum);
if (_totalWritten == 0)
_firstSequence = sequence;
@@ -274,6 +300,11 @@ public sealed class MsgBlock : IDisposable
_cache ??= new Dictionary<ulong, MessageRecord>();
_cache[sequence] = record;
// Go: msgBlock.lchk — capture checksum (last 8 bytes of encoded record).
// Reference: golang/nats-server/server/filestore.go:2204 (lchk update on write)
_lastChecksum ??= new byte[8];
encoded.AsSpan(^8..).CopyTo(_lastChecksum);
if (_totalWritten == 0)
_firstSequence = sequence;
@@ -408,6 +439,11 @@ public sealed class MsgBlock : IDisposable
_skipSequences.Add(sequence); // Track skip sequences separately for recovery
// Note: intentionally NOT added to _cache since it is deleted.
// Go: msgBlock.lchk — capture checksum (last 8 bytes of encoded record).
// Reference: golang/nats-server/server/filestore.go:2204 (lchk update on write)
_lastChecksum ??= new byte[8];
encoded.AsSpan(^8..).CopyTo(_lastChecksum);
if (_totalWritten == 0)
_firstSequence = sequence;