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

Merge branch 'codex/filestore-payload-index-optimization'

Browse Source
This commit is contained in:
Joseph Doherty
2026-03-13 11:36:15 -04:00
parent a1fc600d84 ca2d8019a1
commit 20f45b2aaf
12 changed files with 504 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs
+2 -8
View File
@@ -95,9 +95,6 @@ public sealed class CompiledFilter
public sealed class PullConsumerEngine
{
// Reusable fetch buffer to avoid per-fetch List allocation.
// Go reference: consumer.go — reuses slice backing array across fetch calls.
[ThreadStatic] private static List<StoredMessage>? t_fetchBuf;
// Go: consumer.go — cluster-wide pending pull request tracking keyed by reply subject.
// Reference: golang/nats-server/server/consumer.go waitingRequestsPending / proposeWaitingRequest
private readonly ConcurrentDictionary<string, PullWaitingRequest> _clusterPending =
@@ -158,11 +155,7 @@ public sealed class PullConsumerEngine
public async ValueTask<PullFetchBatch> FetchAsync(StreamHandle stream, ConsumerHandle consumer, PullFetchRequest request, CancellationToken ct)
{
var batch = Math.Max(request.Batch, 1);
// Use thread-static buffer to avoid per-fetch List allocation.
// Results are snapshot'd into PullFetchBatch before the buffer is reused.
var messages = t_fetchBuf ??= new List<StoredMessage>();
messages.Clear();
if (messages.Capacity < batch) messages.Capacity = batch;
var messages = new List<StoredMessage>(batch);
// Go: consumer.go — enforce ExpiresMs timeout on pull fetch requests.
// When ExpiresMs > 0, create a linked CancellationTokenSource that fires
@@ -313,6 +306,7 @@ public sealed class PullConsumerEngine
await Task.Delay(5, ct).ConfigureAwait(false);
}
ct.ThrowIfCancellationRequested();
return null;
}
src/NATS.Server/JetStream/Storage/FileStore.cs
+230 -195
View File
@@ -28,6 +28,8 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
// In-memory cache: keyed by sequence number. This is the primary data structure
// for reads and queries. The blocks are the on-disk persistence layer.
private readonly Dictionary<ulong, StoredMessage> _messages = new();
private readonly Dictionary<ulong, StoredMessageIndex> _messageIndexes = new();
private readonly Dictionary<string, ulong> _lastSequenceBySubject = new(StringComparer.Ordinal);
// Block-based storage: the active (writable) block and sealed blocks.
private readonly List<MsgBlock> _blocks = [];
@@ -141,20 +143,15 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
// We keep _messages for LoadAsync/RemoveAsync but avoid double payload.ToArray().
var persistedPayload = TransformForPersist(payload.Span);
var storedPayload = _noTransform ? persistedPayload : payload.ToArray();
_messages[_last] = new StoredMessage
TrackMessage(new StoredMessage
{
Sequence = _last,
Subject = subject,
Payload = storedPayload,
TimestampUtc = now,
};
});
_generation++;
_messageCount++;
_totalBytes += (ulong)payload.Length;
if (_messageCount == 1)
_firstSeq = _last;
// Go: register TTL only when TTL > 0.
if (_options.MaxAgeMs > 0)
RegisterTtl(_last, timestamp, (long)_options.MaxAgeMs * 1_000_000L);
@@ -195,11 +192,13 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
public ValueTask<StoredMessage?> LoadLastBySubjectAsync(string subject, CancellationToken ct)
{
var match = _messages.Values
.Where(m => string.Equals(m.Subject, subject, StringComparison.Ordinal))
.OrderByDescending(m => m.Sequence)
.FirstOrDefault();
return ValueTask.FromResult(match);
if (_lastSequenceBySubject.TryGetValue(subject, out var sequence)
&& _messages.TryGetValue(sequence, out var match))
{
return ValueTask.FromResult<StoredMessage?>(match);
}
return ValueTask.FromResult<StoredMessage?>(null);
}
public ValueTask<IReadOnlyList<StoredMessage>> ListAsync(CancellationToken ct)
@@ -212,21 +211,11 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
public ValueTask<bool> RemoveAsync(ulong sequence, CancellationToken ct)
{
if (!_messages.TryGetValue(sequence, out var msg))
if (!RemoveTrackedMessage(sequence, preserveHighWaterMark: false))
return ValueTask.FromResult(false);
_messages.Remove(sequence);
_generation++;
// Incremental state tracking.
_messageCount--;
_totalBytes -= (ulong)msg.Payload.Length;
if (sequence == _firstSeq)
_firstSeq = _messages.Count == 0 ? 0UL : _messages.Keys.Min();
if (sequence == _last)
_last = _messages.Count == 0 ? 0UL : _messages.Keys.Max();
// Soft-delete in the block that contains this sequence.
DeleteInBlock(sequence);
@@ -266,6 +255,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
Sequence = x.Sequence,
Subject = x.Subject,
HeadersBase64 = x.RawHeaders.IsEmpty ? null : Convert.ToBase64String(x.RawHeaders.Span),
PayloadBase64 = Convert.ToBase64String(TransformForPersist(x.Payload.Span)),
TimestampUtc = x.TimestampUtc,
})
@@ -276,10 +266,13 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
public ValueTask RestoreSnapshotAsync(ReadOnlyMemory<byte> snapshot, CancellationToken ct)
{
_messages.Clear();
_messageIndexes.Clear();
_lastSequenceBySubject.Clear();
_last = 0;
_messageCount = 0;
_totalBytes = 0;
_firstSeq = 0;
_first = 0;
// Dispose existing blocks and clean files.
DisposeAllBlocks();
@@ -292,25 +285,26 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
foreach (var record in records)
{
var restoredHeaders = string.IsNullOrEmpty(record.HeadersBase64)
? ReadOnlyMemory<byte>.Empty
: Convert.FromBase64String(record.HeadersBase64);
var restoredPayload = RestorePayload(Convert.FromBase64String(record.PayloadBase64 ?? string.Empty));
var message = new StoredMessage
{
Sequence = record.Sequence,
Subject = record.Subject ?? string.Empty,
RawHeaders = restoredHeaders,
Payload = restoredPayload,
TimestampUtc = record.TimestampUtc,
};
_messages[record.Sequence] = message;
_last = Math.Max(_last, record.Sequence);
}
// Recompute incremental state from restored messages.
_messageCount = (ulong)_messages.Count;
_totalBytes = (ulong)_messages.Values.Sum(m => (long)m.Payload.Length);
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
}
}
RebuildIndexesFromMessages();
// Write all messages to fresh blocks.
RewriteBlocks();
return ValueTask.CompletedTask;
@@ -332,14 +326,10 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
var trimmed = false;
while ((ulong)_messages.Count > maxMessages)
{
var first = _messages.Keys.Min();
if (_messages.TryGetValue(first, out var msg))
{
_totalBytes -= (ulong)msg.Payload.Length;
_messageCount--;
}
var first = _firstSeq;
if (first == 0 || !RemoveTrackedMessage(first, preserveHighWaterMark: true))
break;
_messages.Remove(first);
DeleteInBlock(first);
trimmed = true;
}
@@ -347,7 +337,6 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
if (!trimmed)
return;
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
_generation++;
}
@@ -377,36 +366,19 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
var now = DateTime.UtcNow;
var timestamp = new DateTimeOffset(now).ToUnixTimeMilliseconds() * 1_000_000L;
// Combine headers and payload (headers precede the body in NATS wire format).
byte[] combined;
if (hdr is { Length: > 0 })
{
combined = new byte[hdr.Length + (msg?.Length ?? 0)];
hdr.CopyTo(combined, 0);
msg?.CopyTo(combined, hdr.Length);
}
else
{
combined = msg ?? [];
}
var persistedPayload = TransformForPersist(combined.AsSpan());
var stored = new StoredMessage
var headers = hdr is { Length: > 0 } ? hdr : [];
var payload = msg ?? [];
var persistedPayload = TransformForPersist(payload);
TrackMessage(new StoredMessage
{
Sequence = _last,
Subject = subject,
Payload = combined,
RawHeaders = headers,
Payload = payload,
TimestampUtc = now,
};
_messages[_last] = stored;
});
_generation++;
// Incremental state tracking.
_messageCount++;
_totalBytes += (ulong)combined.Length;
if (_messageCount == 1)
_firstSeq = _last;
// Determine effective TTL: per-message ttl (ns) takes priority over MaxAgeMs.
// Go: filestore.go:6830 — if msg.ttl > 0 use it, else use cfg.MaxAge.
var effectiveTtlNs = ttl > 0 ? ttl : (_options.MaxAgeMs > 0 ? (long)_options.MaxAgeMs * 1_000_000L : 0L);
@@ -415,16 +387,16 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
EnsureActiveBlock();
try
{
_activeBlock!.WriteAt(_last, subject, ReadOnlyMemory<byte>.Empty, persistedPayload, timestamp);
_activeBlock!.WriteAt(_last, subject, headers, persistedPayload, timestamp);
}
catch (InvalidOperationException)
{
RotateBlock();
_activeBlock!.WriteAt(_last, subject, ReadOnlyMemory<byte>.Empty, persistedPayload, timestamp);
_activeBlock!.WriteAt(_last, subject, headers, persistedPayload, timestamp);
}
// Go: filestore.go:4443 (setupWriteCache) — record write in bounded cache manager.
_writeCache.TrackWrite(_activeBlock!.BlockId, persistedPayload.Length);
_writeCache.TrackWrite(_activeBlock!.BlockId, headers.Length + persistedPayload.Length);
// Signal the background flush loop to coalesce and flush pending writes.
_flushSignal.Writer.TryWrite(0);
@@ -443,11 +415,14 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
var count = (ulong)_messages.Count;
_messages.Clear();
_messageIndexes.Clear();
_lastSequenceBySubject.Clear();
_generation++;
_last = 0;
_messageCount = 0;
_totalBytes = 0;
_firstSeq = 0;
_first = 0;
DisposeAllBlocks();
CleanBlockFiles();
@@ -470,42 +445,53 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
if (string.IsNullOrEmpty(subject) && keep == 0 && seq == 0)
return Purge();
// Collect all messages matching the subject (with wildcard support) at or below seq, ordered by sequence.
var candidates = _messages.Values
.Where(m => SubjectMatchesFilter(m.Subject, subject))
.Where(m => seq == 0 || m.Sequence <= seq)
.OrderBy(m => m.Sequence)
.ToList();
if (candidates.Count == 0)
var upperBound = seq == 0 ? _last : Math.Min(seq, _last);
if (upperBound == 0)
return 0;
// Keep the newest `keep` messages; purge the rest.
var toRemove = keep > 0 && (ulong)candidates.Count > keep
? candidates.Take(candidates.Count - (int)keep).ToList()
: (keep == 0 ? candidates : []);
if (toRemove.Count == 0)
return 0;
foreach (var msg in toRemove)
ulong candidateCount = 0;
for (var current = _messageCount == 0 ? 0UL : _first; current != 0 && current <= upperBound; current++)
{
_messages.Remove(msg.Sequence);
_totalBytes -= (ulong)msg.Payload.Length;
_messageCount--;
DeleteInBlock(msg.Sequence);
if (!_messages.TryGetValue(current, out var message))
continue;
if (!SubjectMatchesFilter(message.Subject, subject))
continue;
candidateCount++;
}
if (candidateCount == 0)
return 0;
var targetRemoveCount = keep > 0
? (candidateCount > keep ? candidateCount - keep : 0UL)
: candidateCount;
if (targetRemoveCount == 0)
return 0;
ulong removed = 0;
for (var current = _messageCount == 0 ? 0UL : _first; current != 0 && current <= upperBound && removed < targetRemoveCount; current++)
{
if (!_messages.TryGetValue(current, out var message))
continue;
if (!SubjectMatchesFilter(message.Subject, subject))
continue;
if (RemoveTrackedMessage(current, preserveHighWaterMark: true))
{
DeleteInBlock(current);
removed++;
}
}
if (removed == 0)
return 0;
_generation++;
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
// Update _last if required.
if (_messages.Count == 0)
_last = 0;
else if (!_messages.ContainsKey(_last))
_last = _messages.Keys.Max();
return (ulong)toRemove.Count;
return removed;
}
/// <summary>
@@ -524,13 +510,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
foreach (var s in toRemove)
{
if (_messages.TryGetValue(s, out var msg))
{
_totalBytes -= (ulong)msg.Payload.Length;
_messageCount--;
}
_messages.Remove(s);
RemoveTrackedMessage(s, preserveHighWaterMark: true);
DeleteInBlock(s);
}
@@ -545,11 +525,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
}
else
{
if (!_messages.ContainsKey(_last))
_last = _messages.Keys.Max();
// Update _first to reflect the real first message.
_first = _messages.Keys.Min();
_firstSeq = _first;
_first = _firstSeq;
}
return (ulong)toRemove.Length;
@@ -566,11 +542,14 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
// Truncate to nothing.
_messages.Clear();
_messageIndexes.Clear();
_lastSequenceBySubject.Clear();
_generation++;
_last = 0;
_messageCount = 0;
_totalBytes = 0;
_firstSeq = 0;
_first = 0;
DisposeAllBlocks();
CleanBlockFiles();
return;
@@ -579,13 +558,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
var toRemove = _messages.Keys.Where(k => k > seq).ToArray();
foreach (var s in toRemove)
{
if (_messages.TryGetValue(s, out var msg))
{
_totalBytes -= (ulong)msg.Payload.Length;
_messageCount--;
}
_messages.Remove(s);
RemoveTrackedMessage(s, preserveHighWaterMark: false);
DeleteInBlock(s);
}
@@ -594,8 +567,12 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
// Update _last to the new highest existing sequence (or seq if it exists,
// or the highest below seq).
_last = _messages.Count == 0 ? 0 : _messages.Keys.Max();
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
if (_messageCount == 0)
{
_last = 0;
_first = 0;
_firstSeq = 0;
}
}
/// <summary>
@@ -865,6 +842,119 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
}
}
private void TrackMessage(StoredMessage message)
{
_messages[message.Sequence] = message;
_messageIndexes[message.Sequence] = message.ToIndex();
_lastSequenceBySubject[message.Subject] = message.Sequence;
_messageCount++;
_totalBytes += (ulong)(message.RawHeaders.Length + message.Payload.Length);
if (_messageCount == 1)
{
_first = message.Sequence;
_firstSeq = message.Sequence;
}
}
private bool RemoveTrackedMessage(ulong sequence, bool preserveHighWaterMark)
{
if (!_messages.Remove(sequence, out var message))
return false;
_messageIndexes.Remove(sequence);
_messageCount--;
_totalBytes -= (ulong)(message.RawHeaders.Length + message.Payload.Length);
UpdateLastSequenceForSubject(message.Subject, sequence);
if (_messageCount == 0)
{
if (preserveHighWaterMark)
_first = _last + 1;
else
_last = 0;
_firstSeq = 0;
return true;
}
if (sequence == _firstSeq)
AdvanceFirstSequence(sequence + 1);
if (!preserveHighWaterMark && sequence == _last)
_last = FindPreviousLiveSequence(sequence);
return true;
}
private void AdvanceFirstSequence(ulong start)
{
var candidate = start;
while (!_messageIndexes.ContainsKey(candidate) && candidate <= _last)
candidate++;
if (candidate <= _last)
{
_first = candidate;
_firstSeq = candidate;
return;
}
_first = _last + 1;
_firstSeq = 0;
}
private ulong FindPreviousLiveSequence(ulong startExclusive)
{
if (_messageCount == 0 || startExclusive == 0)
return 0;
for (var seq = startExclusive - 1; ; seq--)
{
if (_messageIndexes.ContainsKey(seq))
return seq;
if (seq == 0)
return 0;
}
}
private void UpdateLastSequenceForSubject(string subject, ulong removedSequence)
{
if (!_lastSequenceBySubject.TryGetValue(subject, out var currentLast) || currentLast != removedSequence)
return;
for (var seq = removedSequence - 1; ; seq--)
{
if (_messageIndexes.TryGetValue(seq, out var candidate) && string.Equals(candidate.Subject, subject, StringComparison.Ordinal))
{
_lastSequenceBySubject[subject] = seq;
return;
}
if (seq == 0)
break;
}
_lastSequenceBySubject.Remove(subject);
}
private void RebuildIndexesFromMessages()
{
_messageIndexes.Clear();
_lastSequenceBySubject.Clear();
_messageCount = 0;
_totalBytes = 0;
_firstSeq = 0;
_first = 0;
foreach (var message in _messages.OrderBy(kv => kv.Key).Select(kv => kv.Value))
TrackMessage(message);
if (_messageCount == 0 && _last > 0)
_first = _last + 1;
}
// -------------------------------------------------------------------------
// Subject matching helper
// -------------------------------------------------------------------------
@@ -1056,9 +1146,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
CleanBlockFiles();
_last = _messages.Count == 0 ? 0UL : _messages.Keys.Max();
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
_messageCount = (ulong)_messages.Count;
_totalBytes = (ulong)_messages.Values.Sum(m => (long)m.Payload.Length);
RebuildIndexesFromMessages();
foreach (var message in _messages.OrderBy(kv => kv.Key).Select(kv => kv.Value))
{
@@ -1068,12 +1156,12 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
EnsureActiveBlock();
try
{
_activeBlock!.WriteAt(message.Sequence, message.Subject, ReadOnlyMemory<byte>.Empty, persistedPayload, timestamp);
_activeBlock!.WriteAt(message.Sequence, message.Subject, message.RawHeaders, persistedPayload, timestamp);
}
catch (InvalidOperationException)
{
RotateBlock();
_activeBlock!.WriteAt(message.Sequence, message.Subject, ReadOnlyMemory<byte>.Empty, persistedPayload, timestamp);
_activeBlock!.WriteAt(message.Sequence, message.Subject, message.RawHeaders, persistedPayload, timestamp);
}
if (_activeBlock!.IsSealed)
@@ -1168,15 +1256,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
}
// Sync _first from _messages; if empty, set to _last+1 (watermark).
if (_messages.Count > 0)
_first = _messages.Keys.Min();
else if (_last > 0)
_first = _last + 1;
// Recompute incremental state from recovered messages.
_messageCount = (ulong)_messages.Count;
_totalBytes = (ulong)_messages.Values.Sum(m => (long)m.Payload.Length);
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
RebuildIndexesFromMessages();
}
/// <summary>
@@ -1207,6 +1287,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
Sequence = record.Sequence,
Subject = record.Subject,
RawHeaders = record.Headers,
Payload = originalPayload,
TimestampUtc = DateTimeOffset.FromUnixTimeMilliseconds(record.Timestamp / 1_000_000L).UtcDateTime,
};
@@ -1370,20 +1451,10 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
// Reference: golang/nats-server/server/filestore.go:expireMsgs — dmap-based removal.
foreach (var seq in expired)
{
if (_messages.Remove(seq, out var msg))
{
_messageCount--;
_totalBytes -= (ulong)msg.Payload.Length;
}
RemoveTrackedMessage(seq, preserveHighWaterMark: true);
DeleteInBlock(seq);
}
if (_messages.Count > 0)
_firstSeq = _messages.Keys.Min();
else
_firstSeq = 0;
_generation++;
}
@@ -1407,14 +1478,8 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
foreach (var sequence in expired)
{
if (_messages.Remove(sequence, out var msg))
{
_messageCount--;
_totalBytes -= (ulong)msg.Payload.Length;
}
RemoveTrackedMessage(sequence, preserveHighWaterMark: true);
}
_firstSeq = _messages.Count > 0 ? _messages.Keys.Min() : 0UL;
RewriteBlocks();
}
@@ -1676,27 +1741,13 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
/// </summary>
public bool RemoveMsg(ulong seq)
{
if (!_messages.Remove(seq, out var msg))
if (!RemoveTrackedMessage(seq, preserveHighWaterMark: true))
return false;
_generation++;
_messageCount--;
_totalBytes -= (ulong)msg.Payload.Length;
// Go: filestore.go — LastSeq (lmb.last.seq) is a high-water mark and is
// never decremented on removal. Only FirstSeq advances when the first
// live message is removed.
if (_messages.Count == 0)
{
_first = _last + 1; // All gone — next first would be after last
_firstSeq = 0;
}
else
{
_first = _messages.Keys.Min();
_firstSeq = _first;
}
DeleteInBlock(seq);
return true;
}
@@ -1709,23 +1760,10 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
/// </summary>
public bool EraseMsg(ulong seq)
{
if (!_messages.Remove(seq, out var msg))
if (!RemoveTrackedMessage(seq, preserveHighWaterMark: true))
return false;
_generation++;
_messageCount--;
_totalBytes -= (ulong)msg.Payload.Length;
if (_messages.Count == 0)
{
_first = _last + 1;
_firstSeq = 0;
}
else
{
_first = _messages.Keys.Min();
_firstSeq = _first;
}
// Secure erase: overwrite payload bytes with random data before marking deleted.
// Reference: golang/nats-server/server/filestore.go:5890 (eraseMsg).
@@ -1813,6 +1851,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
sm ??= new StoreMsg();
sm.Clear();
sm.Subject = stored.Subject;
sm.Header = stored.RawHeaders.Length > 0 ? stored.RawHeaders.ToArray() : null;
sm.Data = stored.Payload.Length > 0 ? stored.Payload.ToArray() : null;
sm.Sequence = stored.Sequence;
sm.Timestamp = new DateTimeOffset(stored.TimestampUtc).ToUnixTimeMilliseconds() * 1_000_000L;
@@ -1833,7 +1872,9 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
sm ??= new StoreMsg();
sm.Clear();
sm.Subject = record.Subject;
sm.Data = record.Payload.Length > 0 ? record.Payload.ToArray() : null;
sm.Header = record.Headers.Length > 0 ? record.Headers.ToArray() : null;
var originalPayload = RestorePayload(record.Payload.Span);
sm.Data = originalPayload.Length > 0 ? originalPayload : null;
sm.Sequence = record.Sequence;
sm.Timestamp = record.Timestamp;
return sm;
@@ -1887,6 +1928,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
sm ??= new StoreMsg();
sm.Clear();
sm.Subject = match.Subject;
sm.Header = match.RawHeaders.Length > 0 ? match.RawHeaders.ToArray() : null;
sm.Data = match.Payload.Length > 0 ? match.Payload.ToArray() : null;
sm.Sequence = match.Sequence;
sm.Timestamp = new DateTimeOffset(match.TimestampUtc).ToUnixTimeMilliseconds() * 1_000_000L;
@@ -1917,6 +1959,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
sm ??= new StoreMsg();
sm.Clear();
sm.Subject = found.Value.Subject;
sm.Header = found.Value.RawHeaders.Length > 0 ? found.Value.RawHeaders.ToArray() : null;
sm.Data = found.Value.Payload.Length > 0 ? found.Value.Payload.ToArray() : null;
sm.Sequence = found.Key;
sm.Timestamp = new DateTimeOffset(found.Value.TimestampUtc).ToUnixTimeMilliseconds() * 1_000_000L;
@@ -2041,20 +2084,9 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
if (_stopped)
throw new ObjectDisposedException(nameof(FileStore), "Store has been stopped.");
// Combine headers and payload, same as StoreMsg.
byte[] combined;
if (hdr is { Length: > 0 })
{
combined = new byte[hdr.Length + msg.Length];
hdr.CopyTo(combined, 0);
msg.CopyTo(combined, hdr.Length);
}
else
{
combined = msg;
}
var persistedPayload = TransformForPersist(combined.AsSpan());
var headers = hdr is { Length: > 0 } ? hdr : [];
var payload = msg ?? [];
var persistedPayload = TransformForPersist(payload);
// Recover UTC DateTime from caller-supplied Unix nanosecond timestamp.
var storedUtc = DateTimeOffset.FromUnixTimeMilliseconds(ts / 1_000_000L).UtcDateTime;
@@ -2062,10 +2094,11 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
Sequence = seq,
Subject = subject,
Payload = combined,
RawHeaders = headers,
Payload = payload,
TimestampUtc = storedUtc,
};
_messages[seq] = stored;
TrackMessage(stored);
_generation++;
// Go: update _last to the high-water mark — do not decrement.
@@ -2078,16 +2111,16 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
EnsureActiveBlock();
try
{
_activeBlock!.WriteAt(seq, subject, ReadOnlyMemory<byte>.Empty, persistedPayload, ts);
_activeBlock!.WriteAt(seq, subject, headers, persistedPayload, ts);
}
catch (InvalidOperationException)
{
RotateBlock();
_activeBlock!.WriteAt(seq, subject, ReadOnlyMemory<byte>.Empty, persistedPayload, ts);
_activeBlock!.WriteAt(seq, subject, headers, persistedPayload, ts);
}
// Go: filestore.go:4443 (setupWriteCache) — record write in bounded cache manager.
_writeCache.TrackWrite(_activeBlock!.BlockId, persistedPayload.Length);
_writeCache.TrackWrite(_activeBlock!.BlockId, headers.Length + persistedPayload.Length);
if (_activeBlock!.IsSealed)
RotateBlock();
@@ -2113,6 +2146,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
sm ??= new StoreMsg();
sm.Clear();
sm.Subject = stored.Subject;
sm.Header = stored.RawHeaders.Length > 0 ? stored.RawHeaders.ToArray() : null;
sm.Data = stored.Payload.Length > 0 ? stored.Payload.ToArray() : null;
sm.Sequence = stored.Sequence;
sm.Timestamp = new DateTimeOffset(stored.TimestampUtc).ToUnixTimeMilliseconds() * 1_000_000L;
@@ -2365,6 +2399,7 @@ public sealed class FileStore : IStreamStore, IAsyncDisposable, IDisposable
{
public ulong Sequence { get; init; }
public string? Subject { get; init; }
public string? HeadersBase64 { get; init; }
public string? PayloadBase64 { get; init; }
public DateTime TimestampUtc { get; init; }
}
src/NATS.Server/JetStream/Storage/StoredMessage.cs
+4
View File
@@ -5,6 +5,7 @@ public sealed class StoredMessage
public ulong Sequence { get; init; }
public string Subject { get; init; } = string.Empty;
public ReadOnlyMemory<byte> Payload { get; init; }
internal ReadOnlyMemory<byte> RawHeaders { get; init; }
public DateTime TimestampUtc { get; init; } = DateTime.UtcNow;
public string? Account { get; init; }
public bool Redelivered { get; init; }
@@ -18,4 +19,7 @@ public sealed class StoredMessage
/// Convenience accessor for the Nats-Msg-Id header value, used by source deduplication.
/// </summary>
public string? MsgId => Headers is not null && Headers.TryGetValue("Nats-Msg-Id", out var id) ? id : null;
internal StoredMessageIndex ToIndex()
=> new(Sequence, Subject, Payload.Length, TimestampUtc);
}
src/NATS.Server/JetStream/Storage/StoredMessageIndex.cs
+7
View File
@@ -0,0 +1,7 @@
namespace NATS.Server.JetStream.Storage;
public readonly record struct StoredMessageIndex(
ulong Sequence,
string Subject,
int PayloadLength,
DateTime TimestampUtc);