fleet: require a minimum version of the peer at the handshake [CI]
ci/woodpecker/push/web-build Pipeline was successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/push/dev-image Pipeline was successful
ci/woodpecker/push/dev-backend-image Pipeline was successful

MoShiro__MinAgentVersion makes the backend refuse agents below it at hello (fleet_version_required); fleet.min_server_version or MOSHIRO_MIN_SERVER_VERSION makes the agent drop a backend below it. Either side waits five seconds and shakes hands again, because the other one may be updating.
This commit is contained in:
justme committed 2026-10-22 05:28:29 +08:00
1 parent fb3a9c260f
commit 5822285a88
16 files changed
+266 -7

No files matched your search

@@ -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,
@@ -40,6 +40,10 @@ public sealed class FleetSection
/// <summary>节点自己要求有属主(passkey):没有就不运行。见 <c>OwnerGuard</c>。</summary>
[JsonPropertyName("require_owner")]
public bool RequireOwner { get; set; } = true;
/// <summary>面板至少要是哪个版本才继续连接(如 <c>0.4.2-dev.80</c>);空 = 不要求。不够就不发任何东西,5 秒后重新握手(面板可能正在更新)。环境变量 <c>MOSHIRO_MIN_SERVER_VERSION</c> 优先。</summary>
[JsonPropertyName("min_server_version")]
public string? MinServerVersion { get; set; }
}
public sealed class ListenSection
@@ -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");
/// <summary>对方的版本低于要求的最低版本:等一会儿(对方可能正在更新)再握手。</summary>
public static readonly ErrorCode VersionRequired = new("fleet_version_required");
}
@@ -0,0 +1,87 @@
using System.Globalization;
namespace Fleet.Core;
/// <summary>
/// 版本号比较,给「要求的最低版本」用。认得 <c>主.次.修订</c>,可以带预发布后缀;开发构建 <c>0.4.2-dev.73-5245f03</c> 按 <c>dev.73</c> 的序号比较,
/// 提交哈希不参与比较。同一个主次修订里,正式版高于任何预发布版。认不出来的版本号不拒绝(返回 false),免得一个奇怪的版本字符串把节点挡在门外。
/// </summary>
public static class NodeVersions
{
private readonly record struct Parsed(int Major, int Minor, int Patch, bool Prerelease, long Dev, string Suffix);
/// <summary><paramref name="actual"/> 低于 <paramref name="required"/> 返回 true;任何一个为空或无法解析返回 false。</summary>
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;
}
}
+22 -2
View File
@@ -52,6 +52,9 @@ public sealed class FleetLinkOptions
public string NodeVersion { get; init; } = "0.0.0";
/// <summary>服务端至少要是哪个版本(每次握手重新取);空表示不要求。不够就关掉这条连接,等 <see cref="VersionRetry"/> 再握手,服务端可能正在更新。</summary>
public Func<string?>? MinServerVersion { get; init; }
/// <summary>框架之外的能力,例如 <c>moshiro.v1</c> 与 <c>moshiro.kinds:udp_echo,tcp_echo</c>。</summary>
public IReadOnlyList<string> Capabilities { get; init; } = [];
@@ -100,7 +103,11 @@ public sealed class FleetLink
private volatile RpcSession<LinkContext>? _session;
private volatile bool _stopped;
/// <summary>版本不够时重新握手的间隔:固定,不退避(对方更新完马上就能连上)。</summary>
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<LinkContext> router, IEnumerable<IStateHandler> 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> 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();
@@ -104,6 +104,10 @@ public sealed class HelloResult
/// <summary>节点时钟 − 服务端时钟(毫秒,未扣除网络往返)。</summary>
[JsonPropertyName("clock_offset_ms")]
public long ClockOffsetMs { get; set; }
/// <summary>服务端的版本:节点要求了最低服务端版本时,用它判断。</summary>
[JsonPropertyName("server_version")]
public string? ServerVersion { get; set; }
}
public sealed class HeartbeatParams
@@ -22,3 +22,12 @@ public sealed class FleetServerOptions
/// <summary>节点违反协议(方法没有声明、方向、类型或角色不对)的次数上限;超过就断开这条连接。</summary>
public int MaxViolations { get; set; } = 20;
}
/// <summary>握手时的版本要求:服务端的版本(告诉节点),以及节点至少要是哪个版本。每次握手都重新取,所以改配置不用重启。</summary>
public interface IFleetVersionPolicy
{
string? ServerVersion { get; }
/// <summary>空表示不要求。</summary>
string? MinNodeVersion { get; }
}
@@ -24,8 +24,10 @@ public sealed class FleetLinkHost : IFleetConnections, IStateNotifier
IEnumerable<IInbox> inboxes,
IEnumerable<INodeEventSink> 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
/// <summary>服务器的身份密钥;没有时节点连接用旧的明文认证。</summary>
internal IServerIdentity? ServerIdentity { get; }
internal IFleetVersionPolicy? VersionPolicy { get; }
internal FleetServerOptions Options { get; }
internal TimeProvider Time { get; }
@@ -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);
@@ -110,7 +110,8 @@ public static class FleetServerExtensions
provider.GetServices<IInbox>(),
provider.GetServices<INodeEventSink>(),
provider.GetRequiredService<FleetRouter>(),
provider.GetService<IServerIdentity>()));
provider.GetService<IServerIdentity>(),
provider.GetService<IFleetVersionPolicy>()));
services.AddSingleton<IFleetConnections>(provider => provider.GetRequiredService<FleetLinkHost>());
services.AddSingleton<IStateNotifier>(provider => provider.GetRequiredService<FleetLinkHost>());
return builder;
@@ -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);
}
@@ -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));
}
@@ -60,6 +60,9 @@ public sealed class BackendOptions
public int LoginAttemptsPerMinute { get; set; } = 10;
/// <summary>Agent 至少要是哪个版本才能连上(如 <c>0.4.2-dev.80</c>);空 = 不要求。低于它的 Agent 握手时被拒绝,5 秒后重试(它可能正在更新)。环境变量 <c>MoShiro__MinAgentVersion</c>。</summary>
public string? MinAgentVersion { get; set; }
public int RawRetentionDays { get; set; } = 30;
public int TestRawRetentionDays { get; set; } = 90;
@@ -16,6 +16,7 @@ public static class GatewayServiceCollectionExtensions
public static IServiceCollection AddMoShiroGateway(this IServiceCollection services)
{
services.AddSingleton<IAgentGateway, FleetGateway>();
services.AddSingleton<IFleetVersionPolicy, BackendVersionPolicy>();
services.AddHostedService<ConnectionShutdown>();
services.AddFleetEnrollment();
services.AddFleetServer(options => options.NodeIdPrefix = AgentIds.Prefix)
@@ -44,3 +45,11 @@ public static class GatewayServiceCollectionExtensions
(context, parameters, ct) => services.GetRequiredService<TestService>().OnStatusAsync(context.Node.Id, parameters, ct));
}
}
/// <summary>握手时的版本要求:告诉 Agent 面板的版本,并要求 Agent 不低于配置的最低版本。</summary>
internal sealed class BackendVersionPolicy(BackendOptions options) : IFleetVersionPolicy
{
public string? ServerVersion => BuildInfo.Version;
public string? MinNodeVersion => string.IsNullOrWhiteSpace(options.MinAgentVersion) ? null : options.MinAgentVersion;
}
@@ -42,7 +42,7 @@ public sealed class AgentHarness : IAsyncDisposable
public byte[] TlsPublicKey => _identity.Certificate.PublicKey.ExportSubjectPublicKeyInfo();
/// <param name="pin">带信任根固定的加入码(码、创世指纹、属主):节点按固定了创世和属主的方式注册。</param>
public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? configure = null, string? name = null, (string Code, string Genesis, string[] Controllers)? pin = null)
public static async Task<AgentHarness> StartAsync(BackendFixture backend, Action<AgentConfig>? 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;
}
@@ -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<BackendFixture>
{
private readonly BackendFixture _backend;
public VersionRequirementTests(BackendFixture backend)
{
_backend = backend;
}
[Fact]
public async Task AnAgentBelowTheRequiredVersion_IsRejectedAtHello_AndGetsInOnceTheRequirementIsMet()
{
BackendOptions options = _backend.App.Services.GetRequiredService<BackendOptions>();
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<MoShiro.Backend.Application.IAgentGateway>().IsOnline(ok.NodeId));
}
[Fact]
[Trait("Category", "Slow")]
public async Task AnAgentTheBackendRejectedForItsVersion_TriesAgainEveryFiveSeconds_AndComesOnlineWhenAccepted()
{
BackendOptions options = _backend.App.Services.GetRequiredService<BackendOptions>();
IAgentGateway gateway = _backend.App.Services.GetRequiredService<IAgentGateway>();
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<IAgentGateway>();
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));
}
}