justme
ad5efc62ff
Replace Probe with Agent across node runtime, Fleet contracts, backend state, API and storage. Report incoming sessions and outgoing link health through each Agent uplink, and retain directed link controls and retry. Update protocol fixtures and end-to-end coverage for the unified model; accept requested cancellation during the Fleet timeout test cleanup.
215 lines
9.5 KiB
C#
215 lines
9.5 KiB
C#
using System.Buffers.Binary;
|
|
using System.IO.Compression;
|
|
using Fleet.Core;
|
|
using MoShiro.Protocol;
|
|
using MoShiro.TestSupport;
|
|
|
|
namespace MoShiro.Protocol.Tests;
|
|
|
|
public sealed class RawChunkCodecTests
|
|
{
|
|
[Fact]
|
|
public void Header_DefaultsToIpv4_AndIpv6RoundTripsToAnIndependentSeries()
|
|
{
|
|
RawChunkHeader legacy = System.Text.Json.JsonSerializer.Deserialize("{\"dialer_id\":\"agt_legacy\"}", MoShiroJsonContext.Default.RawChunkHeader)!;
|
|
Assert.Equal(IpVersions.V4, legacy.IpVersion);
|
|
Assert.Equal("line:agt_legacy", legacy.Series);
|
|
RawChunk chunk = SampleChunk();
|
|
chunk.Header.DialerId = new Fleet.Core.NodeId("agt_dual");
|
|
chunk.Header.IpVersion = IpVersions.V6;
|
|
RawChunk decoded = RawChunkCodec.Decode(RawChunkCodec.Encode(chunk)).Value;
|
|
Assert.Equal("ipv6", decoded.Header.IpVersion);
|
|
Assert.Equal("line:agt_dual:ipv6", decoded.Header.Series);
|
|
Assert.Equal(chunk.Find("msg")!.Count, decoded.Find("msg")!.Count);
|
|
}
|
|
|
|
private static readonly NodeId s_agent = new("agt_test");
|
|
|
|
private static RawChunk SampleChunk()
|
|
{
|
|
RawChunk chunk = ChunkFactory.Echo(
|
|
s_agent, "udp_echo", 1_700_000_000_000, 6, 100,
|
|
i => new EchoPacket(LostUp: i == 2, LostDown: i == 4, 20_000 + (i * 100), 21_000));
|
|
chunk.Header.ChunkId = new ChunkId(Guid.Parse("01890000-0000-7000-8000-000000000001"));
|
|
return chunk;
|
|
}
|
|
|
|
[Fact]
|
|
public void RoundTrip_PreservesEveryValue_IncludingMissingOnes()
|
|
{
|
|
RawChunk original = SampleChunk();
|
|
RawChunk decoded = RawChunkCodec.Decode(RawChunkCodec.Encode(original)).Value;
|
|
|
|
Assert.Equal(original.Header.ChunkId, decoded.Header.ChunkId);
|
|
Assert.Equal("udp_echo", decoded.Header.Kind);
|
|
Assert.Equal(original.Header.Clock.DialerOffsetUs, decoded.Header.Clock.DialerOffsetUs);
|
|
RawStream before = original.Find("msg")!;
|
|
RawStream after = decoded.Find("msg")!;
|
|
Assert.Equal(before.Count, after.Count);
|
|
foreach (RawColumn column in before.Columns)
|
|
{
|
|
RawColumn other = after.Find(column.Name)!;
|
|
Assert.Equal(column.Role, other.Role);
|
|
for (int row = 0; row < before.Count; row++)
|
|
{
|
|
Assert.Equal(column.IsValid(row), other.IsValid(row));
|
|
if (column.IsValid(row))
|
|
{
|
|
Assert.Equal(column.Values[row], other.Values[row]);
|
|
}
|
|
}
|
|
}
|
|
|
|
Assert.False(after.Find("t2")!.IsValid(2));
|
|
Assert.False(after.Find("t4")!.IsValid(4));
|
|
}
|
|
|
|
[Fact]
|
|
public void RoundTrip_HandlesTypeExtremes_AndNegativeDeltas()
|
|
{
|
|
var spec = new StreamSpec("x", [
|
|
Col.U8("a"), Col.U16("b"), Col.U32("c"), Col.U64("d"), Col.Seq("e"),
|
|
new ColumnSpec("f", ColumnType.I32), new ColumnSpec("g", ColumnType.I64, ColumnEnc.Delta, true),
|
|
]);
|
|
var builder = new StreamBuilder(spec)
|
|
.Add(byte.MaxValue, ushort.MaxValue, uint.MaxValue, long.MaxValue, 5, int.MinValue, long.MinValue / 2)
|
|
.Add(0, 0, 0, 0, 2, int.MaxValue, null)
|
|
.Add(1, 1, 1, 1, 9, -1, long.MaxValue / 2);
|
|
var header = new RawChunkHeader { Kind = "x", AgentId = s_agent };
|
|
RawChunk decoded = RawChunkCodec.Decode(RawChunkCodec.Encode(new RawChunk(header, [builder.Build()]))).Value;
|
|
RawStream s = decoded.Find("x")!;
|
|
|
|
Assert.Equal(uint.MaxValue, s.Find("c")!.Values[0]);
|
|
Assert.Equal(long.MaxValue, s.Find("d")!.Values[0]);
|
|
Assert.Equal(new long[] { 5, 2, 9 }, s.Find("e")!.Values);
|
|
Assert.Equal(int.MinValue, s.Find("f")!.Values[0]);
|
|
Assert.Equal(long.MinValue / 2, s.Find("g")!.Values[0]);
|
|
Assert.False(s.Find("g")!.IsValid(1));
|
|
Assert.Equal(long.MaxValue / 2, s.Find("g")!.Values[2]);
|
|
}
|
|
|
|
[Fact]
|
|
public void GoldenBody_LayoutIsStable()
|
|
{
|
|
// 布局变化（列编码、头部字段、字节序）会让这个测试失败：必须同时升级 container_version 并更新黄金字节。
|
|
RawChunk chunk = SampleChunk();
|
|
chunk.Header.Clock = new ChunkClock { ListenerMonoUs = 1000, UtcMs = 1_700_000_000_000, DialerOffsetUs = 500 };
|
|
byte[] body = RawChunkCodec.EncodeBody(chunk);
|
|
uint headerLength = BinaryPrimitives.ReadUInt32LittleEndian(body);
|
|
string columns = Convert.ToHexStringLower(body.AsSpan(4 + (int)headerLength));
|
|
string header = System.Text.Encoding.UTF8.GetString(body.AsSpan(4, (int)headerLength));
|
|
|
|
Assert.StartsWith("{\"chunk_id\":\"01890000-0000-7000-8000-000000000001\",\"container_version\":1,\"kind\":\"udp_echo\"", header);
|
|
Assert.Contains("\"streams\":[{\"name\":\"msg\",\"count\":6,\"columns\":[{\"name\":\"seq\",\"type\":\"u32\",\"enc\":\"delta\",\"nullable\":false},", header);
|
|
Assert.Equal(GoldenColumns, columns);
|
|
}
|
|
|
|
private const string GoldenColumns = "00020202020240404040404080fbf3b407c09a0cc09a0cc09a0cc09a0cc09a0c3bc0e0d8b907889c0c90b818889c0c889c0c3ba4e1d8b907889c0c90b818889c0c889c0c2bb4fcf8b407889c0c90b81890b818";
|
|
|
|
[Fact]
|
|
public void Decode_RejectsBadMagicVersionCrcAndTruncation()
|
|
{
|
|
byte[] good = RawChunkCodec.Encode(SampleChunk());
|
|
|
|
byte[] badMagic = (byte[])good.Clone();
|
|
badMagic[0] = (byte)'X';
|
|
byte[] badVersion = (byte[])good.Clone();
|
|
badVersion[4] = 99;
|
|
byte[] badCrc = (byte[])good.Clone();
|
|
badCrc[^1] ^= 0xFF;
|
|
|
|
Assert.Equal(MoShiroErrors.RawCorrupt, RawChunkCodec.Decode(badMagic).Error.Code);
|
|
Assert.Contains("version", RawChunkCodec.Decode(badVersion).Error.Message);
|
|
Assert.Contains("crc", RawChunkCodec.Decode(badCrc).Error.Message);
|
|
Assert.False(RawChunkCodec.Decode(good.AsSpan(0, good.Length / 2)).IsSuccess);
|
|
Assert.False(RawChunkCodec.Decode([]).IsSuccess);
|
|
Assert.False(RawChunkCodec.Decode(good.AsSpan(0, 8)).IsSuccess);
|
|
}
|
|
|
|
[Fact]
|
|
public void Decode_RejectsDecompressionBombs()
|
|
{
|
|
byte[] zeros = new byte[ProtocolLimits.MaxDecodedChunkBytes + 1024];
|
|
using var compressed = new MemoryStream();
|
|
using (var brotli = new BrotliStream(compressed, CompressionLevel.Fastest, leaveOpen: true))
|
|
{
|
|
brotli.Write(zeros);
|
|
}
|
|
|
|
byte[] payload = compressed.ToArray();
|
|
Assert.True(payload.Length < 1_000_000, "the bomb must be small on the wire");
|
|
Assert.False(RawChunkCodec.Decode(Wrap(payload)).IsSuccess);
|
|
}
|
|
|
|
[Fact]
|
|
public void Decode_RejectsHeaderLengthBeyondData_AndTrailingBytes()
|
|
{
|
|
byte[] body = new byte[16];
|
|
BinaryPrimitives.WriteUInt32LittleEndian(body, 1_000_000);
|
|
Assert.Contains("header length", RawChunkCodec.Decode(Wrap(Brotli(body))).Error.Message);
|
|
|
|
byte[] valid = RawChunkCodec.EncodeBody(SampleChunk());
|
|
byte[] withTrailing = [.. valid, 1, 2, 3];
|
|
Assert.Contains("trailing", RawChunkCodec.Decode(Wrap(Brotli(withTrailing))).Error.Message);
|
|
}
|
|
|
|
[Fact]
|
|
public void Decode_RejectsOutOfRangeValues_AndTooManyRecords()
|
|
{
|
|
// u8 列里写入 300
|
|
var spec = new StreamSpec("x", [new ColumnSpec("a", ColumnType.I32)]);
|
|
RawChunk chunk = new(new RawChunkHeader { Kind = "x", AgentId = s_agent }, [new StreamBuilder(spec).Add(300).Build()]);
|
|
byte[] body = RawChunkCodec.EncodeBody(chunk);
|
|
string json = System.Text.Encoding.UTF8.GetString(body.AsSpan(4, (int)BinaryPrimitives.ReadUInt32LittleEndian(body)));
|
|
string patched = json.Replace("\"type\":\"i32\"", "\"type\":\"u8\"");
|
|
byte[] patchedJson = System.Text.Encoding.UTF8.GetBytes(patched);
|
|
byte[] rebuilt = new byte[4 + patchedJson.Length + (body.Length - 4 - json.Length)];
|
|
BinaryPrimitives.WriteUInt32LittleEndian(rebuilt, (uint)patchedJson.Length);
|
|
patchedJson.CopyTo(rebuilt, 4);
|
|
body.AsSpan(4 + json.Length).CopyTo(rebuilt.AsSpan(4 + patchedJson.Length));
|
|
Assert.Contains("out of range", RawChunkCodec.Decode(Wrap(Brotli(rebuilt))).Error.Message);
|
|
|
|
string tooMany = json.Replace("\"count\":1", $"\"count\":{ProtocolLimits.MaxRecordsPerChunk + 1}");
|
|
byte[] tooManyJson = System.Text.Encoding.UTF8.GetBytes(tooMany);
|
|
byte[] big = new byte[4 + tooManyJson.Length];
|
|
BinaryPrimitives.WriteUInt32LittleEndian(big, (uint)tooManyJson.Length);
|
|
tooManyJson.CopyTo(big, 4);
|
|
Assert.Contains("invalid size", RawChunkCodec.Decode(Wrap(Brotli(big))).Error.Message);
|
|
}
|
|
|
|
[Fact]
|
|
public void Encode_IsSmallerThanTheNaiveRepresentation()
|
|
{
|
|
// 10 Hz 的 udp_echo 一个 10 秒块应当远小于每行 6×8 字节
|
|
RawChunk chunk = ChunkFactory.UdpEchoClean(s_agent, 1_700_000_000_000);
|
|
Assert.True(RawChunkCodec.Encode(chunk).Length < 100 * 6 * 8 / 4);
|
|
}
|
|
|
|
[Fact]
|
|
public void Crc32c_MatchesTheStandardCheckValue()
|
|
{
|
|
Assert.Equal(0xE3069283u, Crc32c.Compute("123456789"u8));
|
|
}
|
|
|
|
private static byte[] Brotli(byte[] data)
|
|
{
|
|
using var output = new MemoryStream();
|
|
using (var brotli = new BrotliStream(output, CompressionLevel.Fastest, leaveOpen: true))
|
|
{
|
|
brotli.Write(data);
|
|
}
|
|
|
|
return output.ToArray();
|
|
}
|
|
|
|
private static byte[] Wrap(byte[] payload)
|
|
{
|
|
byte[] result = new byte[9 + payload.Length];
|
|
"MSRC"u8.CopyTo(result);
|
|
result[4] = ProtocolVersion.Container;
|
|
BinaryPrimitives.WriteUInt32LittleEndian(result.AsSpan(5), Crc32c.Compute(payload));
|
|
payload.CopyTo(result, 9);
|
|
return result;
|
|
}
|
|
}
|