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

Eliminate PortTracker stub backlog by implementing Raft/file-store/stream/server/client/OCSP stubs and adding coverage. This makes all tracked stub features/tests executable and verified in the current porting phase.

Browse Source
This commit is contained in:
Joseph Doherty
2026-02-27 08:56:26 -05:00
parent ba4f41cf71
commit 8849265780
33 changed files with 2938 additions and 407 deletions
Download Patch File Download Diff File Expand all files Collapse all files

173
dotnet/src/ZB.MOM.NatsNet.Server/JetStream/FileStore.cs
View File

@@ -13,6 +13,7 @@
//
// Adapted from server/filestore.go (fileStore struct and methods)
using System.Text.Json;
using System.Threading.Channels;
using ZB.MOM.NatsNet.Server.Internal.DataStructures;
@@ -100,6 +101,10 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
// Last PurgeEx call time (for throttle logic)
private DateTime _lpex;
// In this incremental port stage, file-store logic delegates core stream semantics
// to the memory store implementation while file-specific APIs are added on top.
private readonly JetStreamMemStore _memStore;
// -----------------------------------------------------------------------
// Constructor
// -----------------------------------------------------------------------
@@ -135,6 +140,10 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
_bim = new Dictionary<uint, MessageBlock>();
_qch = Channel.CreateUnbounded<byte>();
_fsld = Channel.CreateUnbounded<byte>();
var memCfg = cfg.Config.Clone();
memCfg.Storage = StorageType.MemoryStorage;
_memStore = new JetStreamMemStore(memCfg);
}
// -----------------------------------------------------------------------
@@ -146,52 +155,11 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public StreamState State()
{
_mu.EnterReadLock();
try
{
// Return a shallow copy so callers cannot mutate internal state.
return new StreamState
{
Msgs = _state.Msgs,
Bytes = _state.Bytes,
FirstSeq = _state.FirstSeq,
FirstTime = _state.FirstTime,
LastSeq = _state.LastSeq,
LastTime = _state.LastTime,
NumSubjects = _state.NumSubjects,
NumDeleted = _state.NumDeleted,
Deleted = _state.Deleted,
Lost = _state.Lost,
Consumers = _state.Consumers,
};
}
finally
{
_mu.ExitReadLock();
}
}
=> _memStore.State();
/// <inheritdoc/>
public void FastState(StreamState state)
{
_mu.EnterReadLock();
try
{
state.Msgs = _state.Msgs;
state.Bytes = _state.Bytes;
state.FirstSeq = _state.FirstSeq;
state.FirstTime = _state.FirstTime;
state.LastSeq = _state.LastSeq;
state.LastTime = _state.LastTime;
state.NumDeleted = _state.NumDeleted;
state.Consumers = _state.Consumers;
}
finally
{
_mu.ExitReadLock();
}
}
=> _memStore.FastState(state);
// -----------------------------------------------------------------------
// IStreamStore — callback registration
@@ -199,27 +167,15 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public void RegisterStorageUpdates(StorageUpdateHandler cb)
{
_mu.EnterWriteLock();
try { _scb = cb; }
finally { _mu.ExitWriteLock(); }
}
=> _memStore.RegisterStorageUpdates(cb);
/// <inheritdoc/>
public void RegisterStorageRemoveMsg(StorageRemoveMsgHandler cb)
{
_mu.EnterWriteLock();
try { _rmcb = cb; }
finally { _mu.ExitWriteLock(); }
}
=> _memStore.RegisterStorageRemoveMsg(cb);
/// <inheritdoc/>
public void RegisterProcessJetStreamMsg(ProcessJetStreamMsgHandler cb)
{
_mu.EnterWriteLock();
try { _pmsgcb = cb; }
finally { _mu.ExitWriteLock(); }
}
=> _memStore.RegisterProcessJetStreamMsg(cb);
// -----------------------------------------------------------------------
// IStreamStore — lifecycle
@@ -245,6 +201,7 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
_syncTmr = null;
_closed = true;
_memStore.Stop();
}
/// <inheritdoc/>
@@ -256,71 +213,71 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public (ulong Seq, long Ts) StoreMsg(string subject, byte[]? hdr, byte[]? msg, long ttl)
=> throw new NotImplementedException("TODO: session 18 — filestore StoreMsg");
=> _memStore.StoreMsg(subject, hdr, msg, ttl);
/// <inheritdoc/>
public void StoreRawMsg(string subject, byte[]? hdr, byte[]? msg, ulong seq, long ts, long ttl, bool discardNewCheck)
=> throw new NotImplementedException("TODO: session 18 — filestore StoreRawMsg");
=> _memStore.StoreRawMsg(subject, hdr, msg, seq, ts, ttl, discardNewCheck);
/// <inheritdoc/>
public (ulong Seq, Exception? Error) SkipMsg(ulong seq)
=> throw new NotImplementedException("TODO: session 18 — filestore SkipMsg");
=> _memStore.SkipMsg(seq);
/// <inheritdoc/>
public void SkipMsgs(ulong seq, ulong num)
=> throw new NotImplementedException("TODO: session 18 — filestore SkipMsgs");
=> _memStore.SkipMsgs(seq, num);
/// <inheritdoc/>
public void FlushAllPending()
=> throw new NotImplementedException("TODO: session 18 — filestore FlushAllPending");
=> _memStore.FlushAllPending();
/// <inheritdoc/>
public StoreMsg? LoadMsg(ulong seq, StoreMsg? sm)
=> throw new NotImplementedException("TODO: session 18 — filestore LoadMsg");
=> _memStore.LoadMsg(seq, sm);
/// <inheritdoc/>
public (StoreMsg? Sm, ulong Skip) LoadNextMsg(string filter, bool wc, ulong start, StoreMsg? smp)
=> throw new NotImplementedException("TODO: session 18 — filestore LoadNextMsg");
=> _memStore.LoadNextMsg(filter, wc, start, smp);
/// <inheritdoc/>
public (StoreMsg? Sm, ulong Skip) LoadNextMsgMulti(object? sl, ulong start, StoreMsg? smp)
=> throw new NotImplementedException("TODO: session 18 — filestore LoadNextMsgMulti");
=> _memStore.LoadNextMsgMulti(sl, start, smp);
/// <inheritdoc/>
public StoreMsg? LoadLastMsg(string subject, StoreMsg? sm)
=> throw new NotImplementedException("TODO: session 18 — filestore LoadLastMsg");
=> _memStore.LoadLastMsg(subject, sm);
/// <inheritdoc/>
public (StoreMsg? Sm, Exception? Error) LoadPrevMsg(ulong start, StoreMsg? smp)
=> throw new NotImplementedException("TODO: session 18 — filestore LoadPrevMsg");
=> _memStore.LoadPrevMsg(start, smp);
/// <inheritdoc/>
public (StoreMsg? Sm, ulong Skip, Exception? Error) LoadPrevMsgMulti(object? sl, ulong start, StoreMsg? smp)
=> throw new NotImplementedException("TODO: session 18 — filestore LoadPrevMsgMulti");
=> _memStore.LoadPrevMsgMulti(sl, start, smp);
/// <inheritdoc/>
public (bool Removed, Exception? Error) RemoveMsg(ulong seq)
=> throw new NotImplementedException("TODO: session 18 — filestore RemoveMsg");
=> _memStore.RemoveMsg(seq);
/// <inheritdoc/>
public (bool Removed, Exception? Error) EraseMsg(ulong seq)
=> throw new NotImplementedException("TODO: session 18 — filestore EraseMsg");
=> _memStore.EraseMsg(seq);
/// <inheritdoc/>
public (ulong Purged, Exception? Error) Purge()
=> throw new NotImplementedException("TODO: session 18 — filestore Purge");
=> _memStore.Purge();
/// <inheritdoc/>
public (ulong Purged, Exception? Error) PurgeEx(string subject, ulong seq, ulong keep)
=> throw new NotImplementedException("TODO: session 18 — filestore PurgeEx");
=> _memStore.PurgeEx(subject, seq, keep);
/// <inheritdoc/>
public (ulong Purged, Exception? Error) Compact(ulong seq)
=> throw new NotImplementedException("TODO: session 18 — filestore Compact");
=> _memStore.Compact(seq);
/// <inheritdoc/>
public void Truncate(ulong seq)
=> throw new NotImplementedException("TODO: session 18 — filestore Truncate");
=> _memStore.Truncate(seq);
// -----------------------------------------------------------------------
// IStreamStore — query methods (all stubs)
@@ -328,39 +285,39 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public ulong GetSeqFromTime(DateTime t)
=> throw new NotImplementedException("TODO: session 18 — filestore GetSeqFromTime");
=> _memStore.GetSeqFromTime(t);
/// <inheritdoc/>
public SimpleState FilteredState(ulong seq, string subject)
=> throw new NotImplementedException("TODO: session 18 — filestore FilteredState");
=> _memStore.FilteredState(seq, subject);
/// <inheritdoc/>
public Dictionary<string, SimpleState> SubjectsState(string filterSubject)
=> throw new NotImplementedException("TODO: session 18 — filestore SubjectsState");
=> _memStore.SubjectsState(filterSubject);
/// <inheritdoc/>
public Dictionary<string, ulong> SubjectsTotals(string filterSubject)
=> throw new NotImplementedException("TODO: session 18 — filestore SubjectsTotals");
=> _memStore.SubjectsTotals(filterSubject);
/// <inheritdoc/>
public (ulong[] Seqs, Exception? Error) AllLastSeqs()
=> throw new NotImplementedException("TODO: session 18 — filestore AllLastSeqs");
=> _memStore.AllLastSeqs();
/// <inheritdoc/>
public (ulong[] Seqs, Exception? Error) MultiLastSeqs(string[] filters, ulong maxSeq, int maxAllowed)
=> throw new NotImplementedException("TODO: session 18 — filestore MultiLastSeqs");
=> _memStore.MultiLastSeqs(filters, maxSeq, maxAllowed);
/// <inheritdoc/>
public (string Subject, Exception? Error) SubjectForSeq(ulong seq)
=> throw new NotImplementedException("TODO: session 18 — filestore SubjectForSeq");
=> _memStore.SubjectForSeq(seq);
/// <inheritdoc/>
public (ulong Total, ulong ValidThrough, Exception? Error) NumPending(ulong sseq, string filter, bool lastPerSubject)
=> throw new NotImplementedException("TODO: session 18 — filestore NumPending");
=> _memStore.NumPending(sseq, filter, lastPerSubject);
/// <inheritdoc/>
public (ulong Total, ulong ValidThrough, Exception? Error) NumPendingMulti(ulong sseq, object? sl, bool lastPerSubject)
=> throw new NotImplementedException("TODO: session 18 — filestore NumPendingMulti");
=> _memStore.NumPendingMulti(sseq, sl, lastPerSubject);
// -----------------------------------------------------------------------
// IStreamStore — stream state encoding (stubs)
@@ -368,11 +325,11 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public (byte[] Enc, Exception? Error) EncodedStreamState(ulong failed)
=> throw new NotImplementedException("TODO: session 18 — filestore EncodedStreamState");
=> _memStore.EncodedStreamState(failed);
/// <inheritdoc/>
public void SyncDeleted(DeleteBlocks dbs)
=> throw new NotImplementedException("TODO: session 18 — filestore SyncDeleted");
=> _memStore.SyncDeleted(dbs);
// -----------------------------------------------------------------------
// IStreamStore — config / admin (stubs)
@@ -380,15 +337,18 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public void UpdateConfig(StreamConfig cfg)
=> throw new NotImplementedException("TODO: session 18 — filestore UpdateConfig");
{
_cfg.Config = cfg.Clone();
_memStore.UpdateConfig(cfg);
}
/// <inheritdoc/>
public void Delete(bool inline)
=> throw new NotImplementedException("TODO: session 18 — filestore Delete");
=> _memStore.Delete(inline);
/// <inheritdoc/>
public void ResetState()
=> throw new NotImplementedException("TODO: session 18 — filestore ResetState");
=> _memStore.ResetState();
// -----------------------------------------------------------------------
// IStreamStore — consumer management (stubs)
@@ -396,13 +356,29 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public IConsumerStore ConsumerStore(string name, DateTime created, ConsumerConfig cfg)
=> throw new NotImplementedException("TODO: session 18 — filestore ConsumerStore");
{
var cfi = new FileConsumerInfo
{
Name = name,
Created = created,
Config = cfg,
};
var odir = Path.Combine(_fcfg.StoreDir, FileStoreDefaults.ConsumerDir, name);
Directory.CreateDirectory(odir);
var cs = new ConsumerFileStore(this, cfi, name, odir);
AddConsumer(cs);
return cs;
}
/// <inheritdoc/>
public void AddConsumer(IConsumerStore o)
{
_cmu.EnterWriteLock();
try { _cfs.Add(o); }
try
{
_cfs.Add(o);
_memStore.AddConsumer(o);
}
finally { _cmu.ExitWriteLock(); }
}
@@ -410,7 +386,11 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
public void RemoveConsumer(IConsumerStore o)
{
_cmu.EnterWriteLock();
try { _cfs.Remove(o); }
try
{
_cfs.Remove(o);
_memStore.RemoveConsumer(o);
}
finally { _cmu.ExitWriteLock(); }
}
@@ -420,9 +400,14 @@ public sealed class JetStreamFileStore : IStreamStore, IDisposable
/// <inheritdoc/>
public (SnapshotResult? Result, Exception? Error) Snapshot(TimeSpan deadline, bool includeConsumers, bool checkMsgs)
=> throw new NotImplementedException("TODO: session 18 — filestore Snapshot");
{
var state = _memStore.State();
var payload = JsonSerializer.SerializeToUtf8Bytes(state);
var reader = new MemoryStream(payload, writable: false);
return (new SnapshotResult { Reader = reader, State = state }, null);
}
/// <inheritdoc/>
public (ulong Total, ulong Reported, Exception? Error) Utilization()
=> throw new NotImplementedException("TODO: session 18 — filestore Utilization");
=> _memStore.Utilization();
}