From b544d5cccd76735e034c8055918a50927b437474 Mon Sep 17 00:00:00 2001
From: wenxiangchao <wenxiangchao@ihengxin.com.cn>
Date: 星期二, 10 十一月 2026 08:29:04 +0800
Subject: [PATCH] fix: 维护模式重复进入失败与按钮灰死 + Broker 并发验证用例
---
DeviceBroker/BrokerServer.cs | 453 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 453 insertions(+), 0 deletions(-)
diff --git a/DeviceBroker/BrokerServer.cs b/DeviceBroker/BrokerServer.cs
new file mode 100644
index 0000000..5cc103c
--- /dev/null
+++ b/DeviceBroker/BrokerServer.cs
@@ -0,0 +1,453 @@
+using System;
+using System.Buffers.Binary;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Data;
+using System.Globalization;
+using System.IO;
+using System.IO.Pipes;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Linq;
+using Aging.Infrastructure.Broker;
+using Aging.Infrastructure.Contracts;
+using Aging.Infrastructure.Core;
+using Aging.Infrastructure.Drivers;
+using Aging.ResourceContracts;
+using log4net;
+
+namespace DeviceBroker
+{
+ /// <summary>
+ /// DeviceBroker 瀹夸富锛氬敮涓�鎸佹湁 Broker 鍩熻澶囩墿鐞嗕細璇濈殑杩涚▼锛圴3.0 娣峰悎鏋舵瀯锛夈��
+ /// 鍑洪槦鏃跺弻閲嶆牎楠岋細arb.db 鏍呮爮锛坋poch/generation/鐩镐綅锛孖NV-01/03锛�+ configHash 瀵硅处锛圛NV-04锛夛紝
+ /// SessionGate 鎶娾�滈�夋嫨閫氶亾 + 鍐欏叆/鏌ヨ + 璇诲彇鍝嶅簲鈥濊涓轰笉鍙媶鍒嗗師瀛愮墖娈典覆琛屾墽琛岋紙搂04锛夈��
+ /// MVP 涓� OneShot 鍛戒护鍚屾妯″瀷锛氬懆鏈熶换鍔�/璁㈤槄/Outbox 鎺ㄨ繜锛圥2+锛夈��
+ /// </summary>
+ public sealed class BrokerServer
+ {
+ private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerServer));
+
+ private static readonly string[] AllowedOps = { "CONNECT", "DISCONNECT", "SEND", "QUERY", "STATUS" };
+
+ private readonly string _brokerId;
+ private readonly string _pipeName;
+ private readonly string _configDbPath;
+ private readonly DbLeaseStore _lease;
+
+ private sealed class ManagedDevice
+ {
+ public string DeviceId = "";
+ public string Transport = "";
+ public string Endpoint = "";
+ public string SessionKey = "";
+ public IDeviceDriver? Driver;
+ public bool Connected;
+ }
+
+ private readonly Dictionary<string, ManagedDevice> _devices =
+ new(StringComparer.OrdinalIgnoreCase);
+ private readonly ConcurrentDictionary<string, SemaphoreSlim> _gates = new();
+ private readonly SemaphoreSlim _driverCreateLock = new(1, 1);
+
+ private long _hashStampMs;
+ private volatile string _cachedHash = "";
+
+ public BrokerServer(string configDbPath, string arbDbPath, string brokerId)
+ {
+ _configDbPath = configDbPath;
+ _brokerId = brokerId;
+ _pipeName = BrokerPipeName.For(brokerId);
+ _lease = new DbLeaseStore(arbDbPath, brokerId);
+ // 瀹夸富閲嶅惎瀵硅处锛氬洖鏀朵笂娆¤繘绋嬫畫鐣欑殑寰呭洖鏀剁绾︼紙INV-03 鑷姩鍖栧舰寮忥紝璇佹嵁钀藉簱锛�
+ _lease.ReconcileAfterBoot($"broker:{brokerId}/{System.Diagnostics.Process.GetCurrentProcess().Id}");
+ }
+
+ public long Epoch => _lease.CurrentEpoch;
+
+ /// <summary>鍔犺浇鏈� Broker 鍩熻澶囨竻鍗曪紙ExecutionMode=Broker 涓� BrokerId 鍖归厤锛夈��</summary>
+ public void LoadDevices()
+ {
+ _devices.Clear();
+ using var conn = new System.Data.SQLite.SQLiteConnection(
+ $"Data Source={_configDbPath};Version=3;Read Only=True;");
+ conn.Open();
+ using var cmd = conn.CreateCommand();
+ cmd.CommandText = @"SELECT DeviceId, Transport, Endpoint, PhysicalSessionKey, BrokerId, AuthorityId
+ FROM DeviceDefinition WHERE ExecutionMode = 'Broker';";
+ using var rd = cmd.ExecuteReader();
+ var dt = new DataTable();
+ dt.Load(rd);
+ foreach (DataRow row in dt.Rows)
+ {
+ var brokerId = row["BrokerId"].ToString() ?? "";
+ if (!string.Equals(brokerId, _brokerId, StringComparison.OrdinalIgnoreCase))
+ continue; // 灞炲叾浠� Broker 鍩燂紝鏈涓绘嫆缁濅唬鏀�
+ _devices[row["DeviceId"].ToString()!] = new ManagedDevice
+ {
+ DeviceId = row["DeviceId"].ToString() ?? "",
+ Transport = row["Transport"].ToString() ?? "",
+ Endpoint = row["Endpoint"].ToString() ?? "",
+ SessionKey = string.IsNullOrWhiteSpace(row["PhysicalSessionKey"].ToString())
+ ? row["DeviceId"].ToString()! // 鏃犱細璇濋敭鏃堕��鍖栦负鎸夎澶囦覆琛�
+ : row["PhysicalSessionKey"].ToString()!,
+ };
+ }
+ Log.Info($"[Broker] {_brokerId} 杞藉叆 {_devices.Count} 鍙� Broker 璁惧, epoch={_lease.CurrentEpoch}");
+ }
+
+ /// <summary>鎺ュ彈寰幆锛氭瘡涓繛鎺ヤ竴涓鐞嗕换鍔★紱绠￠亾淇濇寔澶氬疄渚嬪彲鐢ㄣ��</summary>
+ public async Task RunAsync(CancellationToken ct)
+ {
+ Log.Info($"[Broker] 鐩戝惉绠￠亾 [{_pipeName}]");
+ while (!ct.IsCancellationRequested)
+ {
+ NamedPipeServerStream? stream = null;
+ try
+ {
+ stream = new NamedPipeServerStream(_pipeName,
+ PipeDirection.InOut, 8, PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
+ await stream.WaitForConnectionAsync(ct).ConfigureAwait(false);
+ var s = stream;
+ stream = null;
+ _ = Task.Run(() => HandleConnectionAsync(s, ct), CancellationToken.None);
+ }
+ catch (OperationCanceledException)
+ {
+ break;
+ }
+ catch (Exception ex)
+ {
+ Log.Error($"[Broker] 鎺ュ彈杩炴帴寮傚父: {ex.Message}", ex);
+ stream?.Dispose();
+ try { await Task.Delay(500, ct).ConfigureAwait(false); }
+ catch (OperationCanceledException) { break; }
+ }
+ }
+ }
+
+ private async Task HandleConnectionAsync(NamedPipeServerStream stream, CancellationToken ct)
+ {
+ var lenBuf = new byte[4];
+ try
+ {
+ using (stream)
+ {
+ while (!ct.IsCancellationRequested && stream.IsConnected)
+ {
+ if (!await ReadExactAsync(stream, lenBuf, 4, ct).ConfigureAwait(false)) break;
+ int len = BinaryPrimitives.ReadInt32BigEndian(lenBuf);
+ if (len <= 0 || len > BrokerIpc.MaxFrameBytes) break;
+ var payload = new byte[len];
+ if (!await ReadExactAsync(stream, payload, len, ct).ConfigureAwait(false)) break;
+
+ IpcEnvelope request;
+ try
+ {
+ request = BrokerIpc.DecodeEnvelope(Encoding.UTF8.GetString(payload));
+ }
+ catch (Exception)
+ {
+ break; // 鍗忚澶辨锛屾柇寮�璇ヨ繛鎺�
+ }
+
+ IpcEnvelope response = await DispatchAsync(request, ct).ConfigureAwait(false);
+ var frame = BrokerIpc.EncodeFrame(BrokerIpc.EncodeEnvelope(response));
+ await stream.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
+ await stream.FlushAsync(ct).ConfigureAwait(false);
+ }
+ }
+ }
+ catch (OperationCanceledException) { }
+ catch (IOException) { }
+ catch (Exception ex)
+ {
+ Log.Warn($"[Broker] 杩炴帴澶勭悊缁撴潫: {ex.Message}");
+ }
+ }
+
+ private async Task<IpcEnvelope> DispatchAsync(IpcEnvelope request, CancellationToken ct)
+ {
+ if (request.ProtocolVersion != BrokerIpc.ProtocolVersion)
+ return Reply(request, BrokerIpc.SerializeBody(new BrokerIpc.ResultBody
+ {
+ State = "Failed",
+ ErrorCode = "PROTOCOL_MISMATCH",
+ Readback = $"Broker 鍗忚 v{BrokerIpc.ProtocolVersion}锛屾嫆缁� v{request.ProtocolVersion}"
+ }));
+
+ switch (request.MessageType)
+ {
+ case IpcMessageTypes.Hello:
+ return Reply(request, BrokerIpc.SerializeBody(new HelloResult
+ {
+ ServerEpoch = _lease.CurrentEpoch,
+ ConfigHash = CurrentHash(),
+ ProtocolVersion = BrokerIpc.ProtocolVersion,
+ BrokerId = _brokerId
+ }));
+
+ case IpcMessageTypes.SubmitTask:
+ var body = BrokerIpc.DeserializeBody<BrokerIpc.SubmitBody>(request.BodyJson);
+ if (body == null)
+ return Reply(request, Result("Failed", "BAD_BODY", "SubmitBody 瑙ｆ瀽澶辫触"));
+ return Reply(request, await ExecuteOneShotAsync(body, request, ct).ConfigureAwait(false));
+
+ case IpcMessageTypes.ProbeReadback:
+ var pb = BrokerIpc.DeserializeBody<BrokerIpc.SubmitBody>(request.BodyJson);
+ return Reply(request, await ProbeReadbackAsync(pb?.DeviceId ?? "").ConfigureAwait(false));
+
+ case IpcMessageTypes.CancelTask:
+ // MVP 涓嶆敮鎸佷腑鏂湪閫斿師瀛愮墖娈碉紙INV-06锛氫笉寰楀亣瑁呭凡鍙栨秷锛�
+ return Reply(request, Result("Failed", "NOT_SUPPORTED", "MVP 涓嶆敮鎸佸彇娑堬紱鐗囨灏嗗湪 deadline 鍐呭畬鎴�"));
+
+ default:
+ return Reply(request, Result("Failed", "UNKNOWN_MESSAGE", request.MessageType));
+ }
+ }
+
+ // ==================== OneShot 鎵ц鏍稿績 ====================
+
+ private async Task<BrokerIpc.ResultBody> ExecuteOneShotAsync(
+ BrokerIpc.SubmitBody body, IpcEnvelope request, CancellationToken ct)
+ {
+ // 0) OpCode 鐧藉悕鍗�
+ if (!AllowedOps.Contains(body.OperationCode) ||
+ !_devices.TryGetValue(body.DeviceId, out var device))
+ {
+ return Result("Failed", "DEVICE_NOT_BROKERED",
+ $"璁惧 {body.DeviceId} 涓嶅睘浜庢湰 Broker 鍩熸垨鎿嶄綔 {body.OperationCode} 涓嶅湪鐧藉悕鍗�");
+ }
+
+ // 1) INV-04锛歝onfigHash 瀵硅处锛堢┖鍝堝笇瑙嗕负鏃у鎴风锛屾斁琛屽苟璁板綍锛�
+ var hash = CurrentHash();
+ if (!string.IsNullOrWhiteSpace(request.ConfigHash) &&
+ !string.Equals(request.ConfigHash, hash, StringComparison.OrdinalIgnoreCase))
+ {
+ return Result("RejectedConfigMismatch", "CONFIG_MISMATCH",
+ $"瀹㈡埛绔揩鐓� {Trim(request.ConfigHash)} 鈮� Broker 鐢熸晥 {Trim(hash)}锛屽厛 Apply 鍐� I/O");
+ }
+
+ // 2) 鏍呮爮鍐嶆牎楠岋細CONNECT/DISCONNECT 浠呯櫥璁版垨閲婃斁瀹夸富鐗╃悊浼氳瘽銆佹棤鎬荤嚎杈撳嚭锛岃眮鍏嶇绾︼紱
+ // 绾鍥炴敹閭锛圧ECV/RECV_ALL/LOG/BUS_STATUS锛変负琚姩鐩戞帶鏃犲啓鍓綔鐢紝涔熻眮鍏嶏紙鏀寔璋冭瘯椤靛彧璇荤洃鍚級锛�
+ // SEND 鍙婂叾瀹� QUERY锛堝彲鑳借Е鍙戣澶囪涓猴紝濡� AUTO_RESPONSE/FILTER锛夊繀椤绘湁娲昏穬绉熺害锛圛NV-01锛�
+ string leaseKey = string.IsNullOrWhiteSpace(body.ResourceKey) ? body.DeviceId : body.ResourceKey;
+ bool readOnlyDrain = body.OperationCode == "QUERY" &&
+ (body.Payload.StartsWith("RECV", StringComparison.OrdinalIgnoreCase) ||
+ body.Payload.StartsWith("LOG", StringComparison.OrdinalIgnoreCase) ||
+ body.Payload.StartsWith("BUS_STATUS", StringComparison.OrdinalIgnoreCase) ||
+ body.Payload.StartsWith("STATUS", StringComparison.OrdinalIgnoreCase));
+ if (body.OperationCode != "DISCONNECT" && body.OperationCode != "CONNECT" && !readOnlyDrain)
+ {
+ var phase = _lease.PhaseOf(leaseKey);
+ if (phase == LeasePhase.RecoveryRequired)
+ {
+ return Result("RejectedStaleLease", "RECOVERY_REQUIRED",
+ $"璧勬簮 {leaseKey} 杩囨湡寰呭洖鏀�(INV-03)锛岀姝� I/O");
+ }
+ var holder = _lease.ProbeConflict(leaseKey);
+ if (phase != LeasePhase.Active || holder == null)
+ {
+ return Result("RejectedStaleLease", "NO_ACTIVE_LEASE",
+ $"璧勬簮 {leaseKey} 鏃犳椿璺冪绾︼紝瀹㈡埛绔』鍏� Acquire锛堢姝㈡棤绉熺害鐩存帴 I/O锛�");
+ }
+ // MVP锛氬悓鏈� arb.db 寮辫韩浠芥牎楠屸�斺�斿彧鍋氬瓨鍦ㄦ��/鐩镐綅妫�鏌ワ紱
+ // owner鈫攃lientInstanceId 绮剧‘瀵归綈寰� epoch 绛惧悕鍑嵁锛圴3.0 搂13 P2锛夈��
+ }
+
+ // 3) SessionGate锛氬悓涓�鐗╃悊浼氳瘽鍘熷瓙鐨勨�滃閬�+鎵ц鈥濈墖娈典覆琛�
+ var gate = _gates.GetOrAdd(device.SessionKey, _ => new SemaphoreSlim(1, 1));
+ bool entered;
+ try { entered = await gate.WaitAsync(5000, CancellationToken.None).ConfigureAwait(false); }
+ catch (ObjectDisposedException) { entered = false; }
+ if (!entered)
+ {
+ // 鏈繘鍏ョ墖娈� 鈫� 纭畾鏈墽琛岋紝鍙畨鍏ㄥ垽蹇�
+ return Result("Failed", "RESOURCE_BUSY",
+ $"浼氳瘽 {device.SessionKey} 鐗囨闃熷垪绛夊緟瓒呮椂锛屾湭鎵ц鍙畨鍏ㄩ噸璇�");
+ }
+
+ try
+ {
+ return await RunFragmentAsync(device, body, ct).ConfigureAwait(false);
+ }
+ finally
+ {
+ gate.Release();
+ }
+ }
+
+ private async Task<BrokerIpc.ResultBody> RunFragmentAsync(
+ ManagedDevice device, BrokerIpc.SubmitBody body, CancellationToken ct)
+ {
+ try
+ {
+ if (body.OperationCode == "CONNECT" || (device.Driver == null && body.OperationCode != "DISCONNECT"))
+ {
+ await EnsureConnectedAsync(device, ct).ConfigureAwait(false);
+ }
+
+ switch (body.OperationCode)
+ {
+ case "CONNECT":
+ return Result("Succeeded", "", $"OK:CONNECTED:{device.DeviceId}", _lease.CurrentEpoch);
+ case "DISCONNECT":
+ if (device.Connected)
+ {
+ await device.Driver!.DisconnectAsync().ConfigureAwait(false);
+ device.Connected = false;
+ }
+ return Result("Succeeded", "", $"OK:DISCONNECTED:{device.DeviceId}", _lease.CurrentEpoch);
+ case "SEND":
+ {
+ EnsureConnectedOrThrow(device);
+ var resp = await device.Driver!.SendCommandAsync(body.Payload, ct).ConfigureAwait(false);
+ return Result("Succeeded", "", AppendPushes(device, resp), _lease.CurrentEpoch);
+ }
+ case "QUERY":
+ {
+ EnsureConnectedOrThrow(device);
+ var resp = await device.Driver!.QueryAsync(body.Payload, ct).ConfigureAwait(false);
+ return Result("Succeeded", "", AppendPushes(device, resp), _lease.CurrentEpoch);
+ }
+ case "STATUS":
+ {
+ var st = device.Driver?.GetStatus();
+ return Result("Succeeded", "",
+ st == null ? "STATUS:NOT_CREATED"
+ : $"STATUS:{st.IsConnected}|{st.LastError}", _lease.CurrentEpoch);
+ }
+ default:
+ return Result("Failed", "BAD_OP", body.OperationCode);
+ }
+ }
+ catch (OperationCanceledException)
+ {
+ // 鐗囨涓�斿彇娑堬細鐗╃悊渚у彲鑳藉凡鎵ц 鈫� INV-06
+ return Result("Unknown", "DEADLINE", $"{body.OperationCode} 瓒� deadline锛岀粨鏋滄湭鐭�");
+ }
+ catch (Exception ex)
+ {
+ bool ioMayHappened = body.OperationCode is "SEND" or "QUERY";
+ return Result(ioMayHappened ? "Unknown" : "Failed", "DRIVER_ERROR",
+ $"{body.OperationCode} 鎵ц寮傚父: {ex.Message}", _lease.CurrentEpoch);
+ }
+ }
+
+ /// <summary>
+ /// 鎹庡甫鎺ㄩ�侊紙搂5.4锛夛細灏嗗涓荤墿鐞嗛┍鍔ㄨ嚜涓婃鍝嶅簲浠ユ潵鏀跺埌鐨� EV:RECV 甯э紝
+ /// 浠� "|PUSHED:n|EV:RECV鈥�" 鍚庣紑鎷煎埌鍝嶅簲 readback锛屽鎴风 BrokerRpcTransport 瑙ｆ瀽涓婂睆銆�
+ /// </summary>
+ private static string AppendPushes(ManagedDevice device, string resp)
+ {
+ if (device.Driver is CanDeviceDriver can)
+ {
+ var lines = can.DrainPendingPushLines();
+ if (lines.Count > 0)
+ return resp + "|PUSHED:" + lines.Count + "|" + string.Join("|", lines);
+ }
+ return resp;
+ }
+
+ private void EnsureConnectedOrThrow(ManagedDevice device)
+ {
+ if (!device.Connected || device.Driver == null)
+ throw new InvalidOperationException($"璁惧 {device.DeviceId} 鏈繛鎺ワ紝璇峰厛 CONNECT");
+ }
+
+ private async Task EnsureConnectedAsync(ManagedDevice device, CancellationToken ct)
+ {
+ if (device.Connected && device.Driver != null) return;
+ await _driverCreateLock.WaitAsync(ct).ConfigureAwait(false);
+ try
+ {
+ if (device.Connected && device.Driver != null) return;
+ device.Driver ??= DriverFactory.CreateDriver(device.DeviceId, device.Transport, device.Endpoint);
+ await device.Driver.ConnectAsync(ct).ConfigureAwait(false);
+ device.Connected = true;
+ Log.Info($"[Broker] 鐗╃悊浼氳瘽宸插缓绔�: {device.DeviceId} ({device.Transport} {device.Endpoint})");
+ }
+ finally
+ {
+ _driverCreateLock.Release();
+ }
+ }
+
+ /// <summary>鎭㈠鍥炴敹鍓嶈鍥炴帰娴嬶細瀹夸富璇诲洖璁惧鐘舵�佷綔涓哄悎鏍艰瘉鎹紙INV-03 浜哄伐鍥炴敹閾剧殑 Broker 渚э級銆�</summary>
+ private async Task<BrokerIpc.ResultBody> ProbeReadbackAsync(string deviceId)
+ {
+ if (!_devices.TryGetValue(deviceId, out var device) || device.Driver == null)
+ return Result("Failed", "NO_DRIVER", $"璁惧 {deviceId} 鏃犳椿鍔ㄩ┍鍔�");
+ try
+ {
+ var readback = await device.Driver.QueryAsync("STATUS", CancellationToken.None).ConfigureAwait(false);
+ return Result("Succeeded", "", readback, _lease.CurrentEpoch);
+ }
+ catch (Exception ex)
+ {
+ return Result("Failed", "READBACK_ERROR", ex.Message);
+ }
+ }
+
+ // ==================== 鏉傞」 ====================
+
+ private string CurrentHash()
+ {
+ long now = BrokerIpc.NowUtcMs();
+ if (now - Interlocked.Read(ref _hashStampMs) > 2000)
+ {
+ try
+ {
+ _cachedHash = DeviceConfigHash.Compute(_configDbPath);
+ }
+ catch (Exception ex)
+ {
+ Log.Warn($"[Broker] configHash 閲嶇畻澶辫触: {ex.Message}");
+ }
+ Interlocked.Exchange(ref _hashStampMs, now);
+ }
+ return _cachedHash;
+ }
+
+ private static BrokerIpc.ResultBody Result(string state, string code, string readback, long epoch = 0)
+ => new BrokerIpc.ResultBody
+ {
+ State = state,
+ ErrorCode = code,
+ Readback = readback,
+ ServerEpoch = epoch
+ };
+
+ private static IpcEnvelope Reply(IpcEnvelope request, string bodyJson) => new IpcEnvelope
+ {
+ ProtocolVersion = BrokerIpc.ProtocolVersion,
+ MessageType = "Result",
+ RequestId = Guid.NewGuid().ToString("N"),
+ CorrelationId = request.RequestId,
+ ClientInstanceId = request.ClientInstanceId,
+ AuthorityId = request.AuthorityId,
+ BodyJson = bodyJson
+ };
+
+ private static IpcEnvelope Reply(IpcEnvelope request, BrokerIpc.ResultBody body)
+ => Reply(request, BrokerIpc.SerializeBody(body));
+
+ private static async Task<bool> ReadExactAsync(Stream s, byte[] buf, int count, CancellationToken ct)
+ {
+ int off = 0;
+ while (off < count)
+ {
+ int n;
+ try { n = await s.ReadAsync(buf, off, count - off, ct).ConfigureAwait(false); }
+ catch (IOException) { return false; }
+ if (n <= 0) return false;
+ off += n;
+ }
+ return true;
+ }
+
+ private static string Trim(string h)
+ => string.IsNullOrEmpty(h) ? "" : (h.Length > 8 ? h.Substring(0, 8) : h);
+ }
+}
--
Gitblit v1.9.1