diff --git a/backend/agent/src/MoShiro.Agent/AgentApp.cs b/backend/agent/src/MoShiro.Agent/AgentApp.cs index 614de02..fa8fc38 100644 --- a/backend/agent/src/MoShiro.Agent/AgentApp.cs +++ b/backend/agent/src/MoShiro.Agent/AgentApp.cs @@ -353,6 +353,7 @@ public sealed class AgentApp : IDisposable, IAsyncDisposable ServerVerifier = _trust, NodeId = _identity.NodeId, NodeVersion = Version, + MinServerVersion = () => Environment.GetEnvironmentVariable("MOSHIRO_MIN_SERVER_VERSION") is { Length: > 0 } required ? required : _config.Fleet.MinServerVersion, Capabilities = [MoShiroCapabilities.Base, MoShiroCapabilities.KindsPrefix + string.Join(',', _listening ? BlockCatalog.AgentKinds : BlockCatalog.TargetKinds)], HelloMeta = HelloMeta, HeartbeatExtra = HeartbeatExtra, diff --git a/backend/agent/src/MoShiro.Agent/AgentConfig.cs b/backend/agent/src/MoShiro.Agent/AgentConfig.cs index 129acb3..16d4447 100644 --- a/backend/agent/src/MoShiro.Agent/AgentConfig.cs +++ b/backend/agent/src/MoShiro.Agent/AgentConfig.cs @@ -40,6 +40,10 @@ public sealed class FleetSection /// 节点自己要求有属主（passkey）：没有就不运行。见 OwnerGuard。 [JsonPropertyName("require_owner")] public bool RequireOwner { get; set; } = true; + + /// 面板至少要是哪个版本才继续连接（如 0.4.2-dev.80）；空 = 不要求。不够就不发任何东西，5 秒后重新握手（面板可能正在更新）。环境变量 MOSHIRO_MIN_SERVER_VERSION 优先。 + [JsonPropertyName("min_server_version")] + public string? MinServerVersion { get; set; } } public sealed class ListenSection diff --git a/backend/fleet/src/Fleet.Core/ErrorCode.cs b/backend/fleet/src/Fleet.Core/ErrorCode.cs index b80e184..b58337e 100644 --- a/backend/fleet/src/Fleet.Core/ErrorCode.cs +++ b/backend/fleet/src/Fleet.Core/ErrorCode.cs @@ -27,4 +27,7 @@ public static class FleetErrors public static readonly ErrorCode JoinTokenExpired = new("fleet_join_token_expired"); public static readonly ErrorCode ProofInvalid = new("fleet_proof_invalid"); public static readonly ErrorCode FrameTooLarge = new("fleet_frame_too_large"); + + /// 对方的版本低于要求的最低版本：等一会儿（对方可能正在更新）再握手。 + public static readonly ErrorCode VersionRequired = new("fleet_version_required"); } diff --git a/backend/fleet/src/Fleet.Core/NodeVersions.cs b/backend/fleet/src/Fleet.Core/NodeVersions.cs new file mode 100644 index 0000000..2e17895 --- /dev/null +++ b/backend/fleet/src/Fleet.Core/NodeVersions.cs @@ -0,0 +1,87 @@ +using System.Globalization; + +namespace Fleet.Core; + +/// +/// 版本号比较，给「要求的最低版本」用。认得 主.次.修订，可以带预发布后缀；开发构建 0.4.2-dev.73-5245f03 按 dev.73 的序号比较， +/// 提交哈希不参与比较。同一个主次修订里，正式版高于任何预发布版。认不出来的版本号不拒绝（返回 false），免得一个奇怪的版本字符串把节点挡在门外。 +/// +public static class NodeVersions +{ + private readonly record struct Parsed(int Major, int Minor, int Patch, bool Prerelease, long Dev, string Suffix); + + /// 低于 返回 true；任何一个为空或无法解析返回 false。 + public static bool IsBelow(string? actual, string? required) + { + if (string.IsNullOrWhiteSpace(required) || string.IsNullOrWhiteSpace(actual)) + { + return false; + } + + if (!TryParse(actual, out Parsed have) || !TryParse(required, out Parsed need)) + { + return false; + } + + return Compare(have, need) < 0; + } + + private static int Compare(Parsed a, Parsed b) + { + int core = (a.Major, a.Minor, a.Patch).CompareTo((b.Major, b.Minor, b.Patch)); + if (core != 0) + { + return core; + } + + if (a.Prerelease != b.Prerelease) + { + return a.Prerelease ? -1 : 1; // 正式版高于预发布版 + } + + if (!a.Prerelease) + { + return 0; + } + + int dev = a.Dev.CompareTo(b.Dev); + return dev != 0 ? dev : 0; + } + + private static bool TryParse(string text, out Parsed parsed) + { + parsed = default; + string value = text.Trim().TrimStart('v'); + int plus = value.IndexOf('+', StringComparison.Ordinal); + if (plus >= 0) + { + value = value[..plus]; + } + + int dash = value.IndexOf('-', StringComparison.Ordinal); + string core = dash < 0 ? value : value[..dash]; + string suffix = dash < 0 ? "" : value[(dash + 1)..]; + string[] parts = core.Split('.'); + if (parts.Length != 3 + || !int.TryParse(parts[0], NumberStyles.None, CultureInfo.InvariantCulture, out int major) + || !int.TryParse(parts[1], NumberStyles.None, CultureInfo.InvariantCulture, out int minor) + || !int.TryParse(parts[2], NumberStyles.None, CultureInfo.InvariantCulture, out int patch)) + { + return false; + } + + long dev = 0; + if (suffix.StartsWith("dev.", StringComparison.Ordinal)) + { + string number = suffix[4..]; + int end = number.IndexOf('-', StringComparison.Ordinal); + if (!long.TryParse(end < 0 ? number : number[..end], NumberStyles.None, CultureInfo.InvariantCulture, out dev)) + { + return false; + } + } + + parsed = new Parsed(major, minor, patch, suffix.Length > 0, dev, suffix); + return true; + } +} diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs index 8b61b19..d6c8248 100644 --- a/backend/fleet/src/Fleet.Link/FleetLink.cs +++ b/backend/fleet/src/Fleet.Link/FleetLink.cs @@ -52,6 +52,9 @@ public sealed class FleetLinkOptions public string NodeVersion { get; init; } = "0.0.0"; + /// 服务端至少要是哪个版本（每次握手重新取）；空表示不要求。不够就关掉这条连接，等 再握手，服务端可能正在更新。 + public Func? MinServerVersion { get; init; } + /// 框架之外的能力，例如 moshiro.v1 与 moshiro.kinds:udp_echo,tcp_echo。 public IReadOnlyList Capabilities { get; init; } = []; @@ -100,7 +103,11 @@ public sealed class FleetLink private volatile RpcSession? _session; private volatile bool _stopped; + /// 版本不够时重新握手的间隔：固定，不退避（对方更新完马上就能连上）。 + public static readonly TimeSpan VersionRetry = TimeSpan.FromSeconds(5); + private static readonly TimeSpan s_tcpTimeout = TimeSpan.FromSeconds(4); + private TimeSpan? _fixedRetry; private int _failedConnects; public FleetLink(FleetLinkOptions options, RpcRouter router, IEnumerable handlers, Outbox outbox, TimeProvider time, INodeLog log) @@ -166,7 +173,8 @@ public sealed class FleetLink return; } - TimeSpan delay = _options.Backoff.Next(); + TimeSpan delay = _fixedRetry ?? _options.Backoff.Next(); + _fixedRetry = null; _log.Warn("link down, retrying", ("err", reason), ("retry_ms", (long)delay.TotalMilliseconds)); await Task.Delay(delay, _time, ct); } @@ -263,7 +271,19 @@ public sealed class FleetLink Result helloResult = await session.RequestAsync(FleetMethods.Hello, hello, FleetRpcContext.Default.HelloParams, FleetRpcContext.Default.HelloResult, ct); if (!helloResult.IsSuccess) { - return "hello failed: " + helloResult.Error.Code; + if (helloResult.Error.Code == FleetErrors.VersionRequired) + { + _fixedRetry = VersionRetry; // 我们的版本低于服务端要求的：等自己更新，每 5 秒再试一次 + } + + return "hello failed: " + helloResult.Error.Code + (_fixedRetry is null ? "" : ": " + helloResult.Error.Message); + } + + string? requiredServer = _options.MinServerVersion?.Invoke(); + if (NodeVersions.IsBelow(helloResult.Value.ServerVersion, requiredServer)) + { + _fixedRetry = VersionRetry; // 服务端版本低于我们要求的（它可能正在更新）：不发任何东西，5 秒后重新握手 + return $"server version {helloResult.Value.ServerVersion} is below the required {requiredServer}"; } Outbox.ResetSent(); diff --git a/backend/fleet/src/Fleet.Rpc/FleetMethods.cs b/backend/fleet/src/Fleet.Rpc/FleetMethods.cs index 4a37c5b..7acc08c 100644 --- a/backend/fleet/src/Fleet.Rpc/FleetMethods.cs +++ b/backend/fleet/src/Fleet.Rpc/FleetMethods.cs @@ -104,6 +104,10 @@ public sealed class HelloResult /// 节点时钟 − 服务端时钟（毫秒，未扣除网络往返）。 [JsonPropertyName("clock_offset_ms")] public long ClockOffsetMs { get; set; } + + /// 服务端的版本：节点要求了最低服务端版本时，用它判断。 + [JsonPropertyName("server_version")] + public string? ServerVersion { get; set; } } public sealed class HeartbeatParams diff --git a/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs b/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs index bca0653..e2122d2 100644 --- a/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs +++ b/backend/fleet/src/Fleet.Server.Abstractions/FleetServerOptions.cs @@ -22,3 +22,12 @@ public sealed class FleetServerOptions /// 节点违反协议（方法没有声明、方向、类型或角色不对）的次数上限；超过就断开这条连接。 public int MaxViolations { get; set; } = 20; } + +/// 握手时的版本要求：服务端的版本（告诉节点），以及节点至少要是哪个版本。每次握手都重新取，所以改配置不用重启。 +public interface IFleetVersionPolicy +{ + string? ServerVersion { get; } + + /// 空表示不要求。 + string? MinNodeVersion { get; } +} diff --git a/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs b/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs index 2f5bd95..5c5646b 100644 --- a/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs +++ b/backend/fleet/src/Fleet.Server.Link/FleetLinkHost.cs @@ -24,8 +24,10 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier IEnumerable inboxes, IEnumerable sinks, FleetRouter router, - IServerIdentity? serverIdentity = null) + IServerIdentity? serverIdentity = null, + IFleetVersionPolicy? versionPolicy = null) { + VersionPolicy = versionPolicy; ServerIdentity = serverIdentity; Store = store; Options = options; @@ -43,6 +45,8 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier /// 服务器的身份密钥；没有时节点连接用旧的明文认证。 internal IServerIdentity? ServerIdentity { get; } + internal IFleetVersionPolicy? VersionPolicy { get; } + internal FleetServerOptions Options { get; } internal TimeProvider Time { get; } diff --git a/backend/fleet/src/Fleet.Server.Link/FleetLog.cs b/backend/fleet/src/Fleet.Server.Link/FleetLog.cs index 10963c9..97e0cec 100644 --- a/backend/fleet/src/Fleet.Server.Link/FleetLog.cs +++ b/backend/fleet/src/Fleet.Server.Link/FleetLog.cs @@ -10,6 +10,9 @@ internal static partial class FleetLog [LoggerMessage(Level = LogLevel.Information, Message = "node disconnected node={Node} reason={Reason}")] public static partial void Disconnected(ILogger logger, string node, string reason); + [LoggerMessage(Level = LogLevel.Warning, Message = "node below the required version node={Node} version={Version} required={Required}")] + public static partial void VersionTooOld(ILogger logger, string node, string version, string required); + [LoggerMessage(Level = LogLevel.Warning, Message = "node auth rejected err={Err}")] public static partial void AuthRejected(ILogger logger, string err); diff --git a/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs b/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs index 4220999..8a7d6e0 100644 --- a/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs +++ b/backend/fleet/src/Fleet.Server.Link/FleetServerBuilder.cs @@ -110,7 +110,8 @@ public static class FleetServerExtensions provider.GetServices(), provider.GetServices(), provider.GetRequiredService(), - provider.GetService())); + provider.GetService(), + provider.GetService())); services.AddSingleton(provider => provider.GetRequiredService()); services.AddSingleton(provider => provider.GetRequiredService()); return builder; diff --git a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs index 333d9a8..3a2d3c6 100644 --- a/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs +++ b/backend/fleet/src/Fleet.Server.Link/NodeConnection.Receive.cs @@ -151,6 +151,20 @@ internal sealed partial class NodeConnection return; } + string? minVersion = _host.VersionPolicy?.MinNodeVersion; + if (NodeVersions.IsBelow(hello.NodeVersion, minVersion)) + { + // 版本不够：不登记在线，告诉节点要求的版本，关闭连接。节点等几秒（它可能正在更新）再握手。 + FleetLog.VersionTooOld(_host.Logger, _node!.Id.Value, hello.NodeVersion, minVersion!); + if (message.Id is not null) + { + await SendAsync(RpcMessage.Failure(message.Id, new FleetError(FleetErrors.VersionRequired, $"node version {hello.NodeVersion} is below the required {minVersion}")), ct); + } + + await CloseAsync("version required"); + return; + } + long nowMs = _host.Time.GetUtcNow().ToUnixTimeMilliseconds(); _session = _session! with { @@ -169,7 +183,7 @@ internal sealed partial class NodeConnection await SendAsync( RpcMessage.Success( message.Id.Value, - new HelloResult { ServerTimeMs = nowMs, ClockOffsetMs = hello.NowUtcMs - nowMs }, + new HelloResult { ServerTimeMs = nowMs, ClockOffsetMs = hello.NowUtcMs - nowMs, ServerVersion = _host.VersionPolicy?.ServerVersion }, FleetRpcContext.Default.HelloResult), ct); } diff --git a/backend/fleet/tests/Fleet.Tests/NodeVersionsTests.cs b/backend/fleet/tests/Fleet.Tests/NodeVersionsTests.cs new file mode 100644 index 0000000..edded19 --- /dev/null +++ b/backend/fleet/tests/Fleet.Tests/NodeVersionsTests.cs @@ -0,0 +1,21 @@ +using Fleet.Core; + +namespace Fleet.Tests; + +public sealed class NodeVersionsTests +{ + [Theory] + [InlineData("0.4.2-dev.73-5245f03", "0.4.2-dev.74", true)] + [InlineData("0.4.2-dev.73-5245f03", "0.4.2-dev.73-abcdef0", false)] // 哈希不参与比较 + [InlineData("0.4.2-dev.100-aaaaaaa", "0.4.2-dev.99", false)] // 序号按数字比，不是按字符串 + [InlineData("0.4.2-dev.999", "0.4.2", true)] // 正式版高于预发布版 + [InlineData("0.4.2", "0.4.2-dev.5", false)] + [InlineData("0.4.1", "0.4.2-dev.1", true)] + [InlineData("v0.5.0", "0.4.9", false)] + [InlineData("0.4.2-dev.73-5245f03", "", false)] // 没有要求 + [InlineData("", "0.4.2", false)] // 认不出来的版本号不拒绝 + [InlineData("not-a-version", "0.4.2", false)] + [InlineData("0.4.2", "garbage", false)] + public void IsBelow_ComparesCoreThenReleaseThenTheDevCounter(string actual, string required, bool expected) => + Assert.Equal(expected, NodeVersions.IsBelow(actual, required)); +} diff --git a/backend/src/MoShiro.Backend.Application/BackendOptions.cs b/backend/src/MoShiro.Backend.Application/BackendOptions.cs index 36e5dc5..603272e 100644 --- a/backend/src/MoShiro.Backend.Application/BackendOptions.cs +++ b/backend/src/MoShiro.Backend.Application/BackendOptions.cs @@ -60,6 +60,9 @@ public sealed class BackendOptions public int LoginAttemptsPerMinute { get; set; } = 10; + /// Agent 至少要是哪个版本才能连上（如 0.4.2-dev.80）；空 = 不要求。低于它的 Agent 握手时被拒绝，5 秒后重试（它可能正在更新）。环境变量 MoShiro__MinAgentVersion。 + public string? MinAgentVersion { get; set; } + public int RawRetentionDays { get; set; } = 30; public int TestRawRetentionDays { get; set; } = 90; diff --git a/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs b/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs index c3bb59c..78867e1 100644 --- a/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs +++ b/backend/src/MoShiro.Backend.Gateway/GatewayServiceCollectionExtensions.cs @@ -16,6 +16,7 @@ public static class GatewayServiceCollectionExtensions public static IServiceCollection AddMoShiroGateway(this IServiceCollection services) { services.AddSingleton(); + services.AddSingleton(); services.AddHostedService(); services.AddFleetEnrollment(); services.AddFleetServer(options => options.NodeIdPrefix = AgentIds.Prefix) @@ -44,3 +45,11 @@ public static class GatewayServiceCollectionExtensions (context, parameters, ct) => services.GetRequiredService().OnStatusAsync(context.Node.Id, parameters, ct)); } } + +/// 握手时的版本要求：告诉 Agent 面板的版本，并要求 Agent 不低于配置的最低版本。 +internal sealed class BackendVersionPolicy(BackendOptions options) : IFleetVersionPolicy +{ + public string? ServerVersion => BuildInfo.Version; + + public string? MinNodeVersion => string.IsNullOrWhiteSpace(options.MinAgentVersion) ? null : options.MinAgentVersion; +} diff --git a/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs b/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs index 271bd35..d82c0af 100644 --- a/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs +++ b/backend/tests/MoShiro.Backend.Tests/AgentHarness.cs @@ -42,7 +42,7 @@ public sealed class AgentHarness : IAsyncDisposable public byte[] TlsPublicKey => _identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo(); /// 带信任根固定的加入码（码、创世指纹、属主）：节点按固定了创世和属主的方式注册。 - public static async Task StartAsync(BackendFixture backend, Action? configure = null, string? name = null, (string Code, string Genesis, string[] Controllers)? pin = null) + public static async Task StartAsync(BackendFixture backend, Action? configure = null, string? name = null, (string Code, string Genesis, string[] Controllers)? pin = null, bool waitForConnection = true) { string directory = Path.Combine(Path.GetTempPath(), "moshiro-agent-" + Guid.NewGuid().ToString("N")); string token = pin?.Code ?? await backend.CreateJoinTokenAsync(); @@ -71,7 +71,11 @@ public sealed class AgentHarness : IAsyncDisposable } await app.Ready; - await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend"); + if (waitForConnection) + { + await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend"); + } + return harness; } diff --git a/backend/tests/MoShiro.Backend.Tests/VersionRequirementTests.cs b/backend/tests/MoShiro.Backend.Tests/VersionRequirementTests.cs new file mode 100644 index 0000000..e3e7ac3 --- /dev/null +++ b/backend/tests/MoShiro.Backend.Tests/VersionRequirementTests.cs @@ -0,0 +1,72 @@ +using System.Text.Json; +using Fleet.Core; +using Fleet.Rpc; +using Microsoft.Extensions.DependencyInjection; +using MoShiro.Backend.Application; +using MoShiro.TestSupport; + +namespace MoShiro.Backend.Tests; + +public sealed class VersionRequirementTests : IClassFixture +{ + private readonly BackendFixture _backend; + + public VersionRequirementTests(BackendFixture backend) + { + _backend = backend; + } + + [Fact] + public async Task AnAgentBelowTheRequiredVersion_IsRejectedAtHello_AndGetsInOnceTheRequirementIsMet() + { + BackendOptions options = _backend.App.Services.GetRequiredService(); + options.MinAgentVersion = "9.0.0"; // 模拟器的版本是 0.0.1-sim + try + { + await using AgentSimulator old = await _backend.ConnectAgentAsync(hello: false); + var hello = new HelloParams { NodeVersion = "0.0.1-sim", NowUtcMs = TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() }; + RpcMessage reply = await old.RequestRawAsync(FleetMethods.Hello, JsonSerializer.SerializeToElement(hello, FleetRpcContext.Default.HelloParams)); + Assert.Equal(FleetErrors.VersionRequired.Value, reply.Error!.Data!.Code); + Assert.Contains("9.0.0", reply.Error.Message); + } + finally + { + options.MinAgentVersion = null; // 共用的后端：恢复，别影响别的测试 + } + + await using AgentSimulator ok = await _backend.ConnectAgentAsync(); + Assert.True(_backend.App.Services.GetRequiredService().IsOnline(ok.NodeId)); + } + + [Fact] + [Trait("Category", "Slow")] + public async Task AnAgentTheBackendRejectedForItsVersion_TriesAgainEveryFiveSeconds_AndComesOnlineWhenAccepted() + { + BackendOptions options = _backend.App.Services.GetRequiredService(); + IAgentGateway gateway = _backend.App.Services.GetRequiredService(); + options.MinAgentVersion = "999.0.0"; + try + { + await using AgentHarness agent = await AgentHarness.StartAsync(_backend, waitForConnection: false); + await Task.Delay(TimeSpan.FromSeconds(2)); + Assert.False(gateway.IsOnline(agent.NodeId)); + + options.MinAgentVersion = null; // 通过了：下一次握手（最迟 5 秒）就上线 + await BackendFixture.EventuallyAsync(() => Task.FromResult(gateway.IsOnline(agent.NodeId)), TimeSpan.FromSeconds(12)); + } + finally + { + options.MinAgentVersion = null; + } + } + + [Fact] + [Trait("Category", "Slow")] + public async Task AnAgentThatRequiresANewerBackend_DoesNotGoOnline_AndKeepsChecking() + { + IAgentGateway gateway = _backend.App.Services.GetRequiredService(); + await using AgentHarness agent = await AgentHarness.StartAsync(_backend, config => config.Fleet.MinServerVersion = "999.0.0", waitForConnection: false); + await Task.Delay(TimeSpan.FromSeconds(7)); // 至少重试过一次 + Assert.False(gateway.IsOnline(agent.NodeId)); + } +}