perf(mtconnect): one Agent write request per device (R06:T-6)

Writes were one POST per slot although the Agent's PUT surface accepts several
id=value pairs. The slots surviving the local refusals now go out as one
WriteObservationsAsync(device, pairs) per device (new IMTConnectAgentClient method;
the default loops the single-pair call). A batch closes at another device or a
repeated DataItem so same-item writes keep request order; a refused multi-slot
request is retried per slot so one refused DataItem costs only its own slot.
Health contract: any Agent answer (accepted or refused) advances
LastSuccessfulRead with the state following the read / stream flags.

Refs #743

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:55:37 -04:00
parent 21b4f00f46
commit f05e3bf6f4
7 changed files with 356 additions and 43 deletions
+12 -2
View File
@@ -419,8 +419,18 @@ step answers a different operator question:
different `level|text` wire format, and are out of scope for v1.
**The Agent's refusal text is logged at Warning and never stored.** It stays out of
`DriverHealth.LastError`, which is type-only by repo-wide discipline, and a refused write
is not evidence the read path degraded — so a refusal leaves driver health untouched.
`DriverHealth.LastError`, which is type-only by repo-wide discipline. A refused write is not
evidence the read path degraded, but it **is** the Agent answering: any answer — accepted or
refused — advances `LastSuccessfulRead`, with the state following the read / `/sample` flags
(a write cannot clear either path's degradation). A write that got **no** answer leaves health
to those two paths, which own liveness.
**One request per device (review 2026-10-21 T-6).** The slots that survive the refusals above
go out as one `POST {agent}/{device}?id1=v1&id2=v2…` per device (the Agent's PUT surface takes
several pairs) instead of one POST per slot. A batch closes at a slot naming another device or
repeating a DataItem already in it, so two writes to one DataItem still land in request order.
An Agent refusal of a multi-slot request is retried slot by slot, so a refused DataItem costs
only its own slot (`BadCommunicationError`); a transport failure fails that request's slots.
**Device resolution needs no extra config.** `TagBinding.WriteTargetsByDataItemId`,
built at bind time from a walk of the cached probe model, remembers which device declared
@@ -188,6 +188,34 @@ public interface IMTConnectAgentClient : IDisposable
Task<MTConnectWriteOutcome> WriteObservationAsync(
string deviceName, string dataItemId, string value, CancellationToken cancellationToken);
/// <summary>
/// Sets several observations on ONE device in one request — the Agent's PUT surface takes any number of
/// <c>{dataItemId}={value}</c> query pairs (<c>POST {agent}/{device}?a=1&amp;b=2</c>) — so a multi-write
/// batch costs one round trip per device instead of one per slot (R06:T-6). Same answer contract as
/// <see cref="WriteObservationAsync"/>: one outcome for the whole request, an Agent refusal is a value, an
/// unreachable Agent throws. The ids must be distinct (a repeated query key's meaning is the Agent's choice).
/// The default issues the single-observation call per pair and reports the first refusal.
/// </summary>
/// <param name="deviceName">The device that owns every DataItem in <paramref name="observations"/>.</param>
/// <param name="observations">The <c>(dataItemId, value)</c> pairs, unescaped, ids distinct.</param>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>What the Agent said about the request.</returns>
async Task<MTConnectWriteOutcome> WriteObservationsAsync(
string deviceName, IReadOnlyList<KeyValuePair<string, string>> observations, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(observations);
foreach (var (dataItemId, value) in observations)
{
var outcome = await WriteObservationAsync(deviceName, dataItemId, value, cancellationToken).ConfigureAwait(false);
if (!outcome.Accepted)
{
return outcome;
}
}
return new MTConnectWriteOutcome(Accepted: true, AgentError: null);
}
/// <summary>
/// Has the Agent's ring buffer overflowed past the sequence the caller expected next? True
/// when the chunk's oldest retained sequence is <i>strictly</i> newer than
@@ -283,8 +283,18 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
}
/// <inheritdoc/>
public async Task<MTConnectWriteOutcome> WriteObservationAsync(
public Task<MTConnectWriteOutcome> WriteObservationAsync(
string deviceName, string dataItemId, string value, CancellationToken cancellationToken)
{
ArgumentException.ThrowIfNullOrWhiteSpace(dataItemId);
ArgumentNullException.ThrowIfNull(value);
return WriteObservationsAsync(deviceName, [new KeyValuePair<string, string>(dataItemId, value)], cancellationToken);
}
/// <inheritdoc/>
public async Task<MTConnectWriteOutcome> WriteObservationsAsync(
string deviceName, IReadOnlyList<KeyValuePair<string, string>> observations, CancellationToken cancellationToken)
{
// Issued on _http, the unary client — the write is a short request/response
// and must be bounded by RequestTimeoutMs like the other unary legs. Sending it on
@@ -293,10 +303,19 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
// watchdog protects a multipart body, not a one-shot POST).
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this);
ArgumentException.ThrowIfNullOrWhiteSpace(deviceName);
ArgumentException.ThrowIfNullOrWhiteSpace(dataItemId);
ArgumentNullException.ThrowIfNull(value);
ArgumentNullException.ThrowIfNull(observations);
if (observations.Count == 0)
{
throw new ArgumentException("at least one observation is required", nameof(observations));
}
var uri = BuildWriteUri(deviceName, dataItemId, value);
foreach (var (dataItemId, value) in observations)
{
ArgumentException.ThrowIfNullOrWhiteSpace(dataItemId, nameof(observations));
ArgumentNullException.ThrowIfNull(value, nameof(observations));
}
var uri = BuildWriteUri(deviceName, observations);
using var lifetime = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _disposeCts.Token);
using var deadline = CancellationTokenSource.CreateLinkedTokenSource(lifetime.Token);
@@ -351,10 +370,11 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable
/// as a query pair. Every caller-supplied part is percent-encoded here — see the seam's
/// contract for why escaping cannot be the caller's job.
/// </summary>
private Uri BuildWriteUri(string deviceName, string dataItemId, string value)
private Uri BuildWriteUri(string deviceName, IReadOnlyList<KeyValuePair<string, string>> observations)
{
var device = Uri.EscapeDataString(deviceName.Trim());
var query = $"{Uri.EscapeDataString(dataItemId)}={Uri.EscapeDataString(value)}";
var query = string.Join(
"&", observations.Select(o => $"{Uri.EscapeDataString(o.Key)}={Uri.EscapeDataString(o.Value)}"));
return new Uri($"{_agentRoot}/{device}?{query}", UriKind.Absolute);
}
@@ -845,9 +845,11 @@ public sealed partial class MTConnectDriver
// form ⇒ BadTypeMismatch; and only then the Agent, whose refusal or whose
// unreachability ⇒ BadCommunicationError.
// The Agent's refusal text is logged, never stored. It goes to a Warning line and
// nowhere near , which is type-only by repo-wide
// discipline — and a refused write is not evidence the read path degraded, so health is
// left exactly as it is (unlike ).
// nowhere near LastError, which is type-only by repo-wide discipline. A refused write is
// not evidence the read path degraded — but it IS the Agent answering, so any answer
// (accepted or refused) advances LastSuccessfulRead with the state following the read /
// stream flags; a write that got no answer leaves health to those paths, which own liveness.
// Slots that survive the local refusals go out as ONE request per device (R06:T-6).
// is null — a caller bug.
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
@@ -868,6 +870,9 @@ public sealed partial class MTConnectDriver
var results = new WriteResult[writes.Count];
// Pass 1 — every local refusal, slot by slot, in the fixed order (see above). What survives is a pending
// Agent write.
var pending = new List<PendingWrite>(writes.Count);
for (var i = 0; i < writes.Count; i++)
{
var request = writes[i];
@@ -903,50 +908,153 @@ public sealed partial class MTConnectDriver
continue;
}
pending.Add(new PendingWrite(i, target.DeviceName, dataItemId, text));
}
if (pending.Count == 0)
{
return results;
}
// Pass 2 — R06:T-6: the Agent's PUT surface takes several id=value pairs per device, so the pending slots go
// out as ONE request per device (it used to be one POST per slot). Request order is kept: a batch closes at
// the first slot naming a different device or repeating an id already in it, so two writes to one DataItem
// land in the order the caller sent them.
var batches = new List<List<PendingWrite>>();
foreach (var write in pending)
{
var open = batches.Count > 0 ? batches[^1] : null;
if (open is null
|| !string.Equals(open[0].DeviceName, write.DeviceName, StringComparison.Ordinal)
|| open.Exists(w => string.Equals(w.DataItemId, write.DataItemId, StringComparison.Ordinal)))
{
batches.Add([write]);
continue;
}
open.Add(write);
}
for (var b = 0; b < batches.Count; b++)
{
var batch = batches[b];
try
{
var outcome = await client
.WriteObservationAsync(target.DeviceName, dataItemId, text, cancellationToken)
.ConfigureAwait(false);
if (outcome.Accepted)
{
results[i] = new WriteResult(0u);
continue;
}
_logger.LogWarning(
"MTConnect driver {DriverInstanceId}: {AgentUri} refused the write of '{Value}' to DataItem '{DataItemId}' on device '{DeviceName}' — {AgentError}. Check the Agent's AllowPut / AllowPutFrom settings and that the DataItem exists.",
_driverInstanceId, options.AgentUri, text, dataItemId, target.DeviceName, outcome.AgentError);
results[i] = new WriteResult(BadCommunicationError);
await WriteBatchAsync(client!, options, batch, results, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// Caller-cancelled: fill the remainder rather than throwing, so the batch shape holds.
for (var j = i; j < writes.Count; j++)
// Caller-cancelled: fill every slot not yet answered rather than throwing, so the batch shape holds
// and a slot that already landed keeps its real outcome.
for (var r = b; r < batches.Count; r++)
{
results[j] = new WriteResult(BadCommunicationError);
foreach (var write in batches[r])
{
results[write.Slot] ??= new WriteResult(BadCommunicationError);
}
}
return results;
}
catch (Exception ex)
{
// A transport failure, a blown per-call deadline, or a disposed client. Logged with the
// exception (its TYPE is the diagnosis) and reported per slot; health is untouched.
_logger.LogWarning(
ex,
"MTConnect driver {DriverInstanceId} could not write DataItem '{DataItemId}' on device '{DeviceName}' to {AgentUri}.",
_driverInstanceId, dataItemId, target.DeviceName, options.AgentUri);
results[i] = new WriteResult(BadCommunicationError);
}
}
return results;
}
/// <summary>One slot that survived the local refusals, waiting for the Agent.</summary>
private sealed record PendingWrite(int Slot, string DeviceName, string DataItemId, string Text);
/// <summary>
/// Sends one same-device batch and maps the Agent's answer onto its slots. Accepted ⇒ every slot Good. Refused
/// ⇒ a single-slot batch is that slot's <c>BadCommunicationError</c> (the Agent's text logged); a multi-slot
/// batch is retried slot by slot so one refused DataItem cannot fail the others — the batch contract above. A
/// transport failure fails the batch's slots. Caller cancellation propagates to <see cref="WriteAsync"/>.
/// </summary>
private async Task WriteBatchAsync(
IMTConnectAgentClient client,
MTConnectDriverOptions options,
List<PendingWrite> batch,
WriteResult[] results,
CancellationToken cancellationToken)
{
var device = batch[0].DeviceName;
MTConnectWriteOutcome outcome;
try
{
outcome = batch.Count == 1
? await client.WriteObservationAsync(device, batch[0].DataItemId, batch[0].Text, cancellationToken)
.ConfigureAwait(false)
: await client.WriteObservationsAsync(
device,
[.. batch.Select(w => new KeyValuePair<string, string>(w.DataItemId, w.Text))],
cancellationToken)
.ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception ex)
{
// A transport failure, a blown per-call deadline, or a disposed client — no answer arrived. Logged with
// the exception (its TYPE is the diagnosis) and reported per slot; health is left to the read and
// stream paths, which own liveness.
_logger.LogWarning(
ex,
"MTConnect driver {DriverInstanceId} could not write {Count} DataItem(s) ({DataItemIds}) on device '{DeviceName}' to {AgentUri}.",
_driverInstanceId, batch.Count, string.Join(", ", batch.Select(w => w.DataItemId)), device, options.AgentUri);
foreach (var write in batch)
{
results[write.Slot] = new WriteResult(BadCommunicationError);
}
return;
}
// An answer — accepted or refused — is the Agent talking (the health contract: a refusal is a reply).
RecordWriteAnswered();
if (outcome.Accepted)
{
foreach (var write in batch)
{
results[write.Slot] = new WriteResult(0u);
}
return;
}
if (batch.Count == 1)
{
_logger.LogWarning(
"MTConnect driver {DriverInstanceId}: {AgentUri} refused the write of '{Value}' to DataItem '{DataItemId}' on device '{DeviceName}' — {AgentError}. Check the Agent's AllowPut / AllowPutFrom settings and that the DataItem exists.",
_driverInstanceId, options.AgentUri, batch[0].Text, batch[0].DataItemId, device, outcome.AgentError);
results[batch[0].Slot] = new WriteResult(BadCommunicationError);
return;
}
// The Agent refused the request as a whole: find out which slot(s) it objects to, one by one.
foreach (var write in batch)
{
await WriteBatchAsync(client, options, [write], results, cancellationToken).ConfigureAwait(false);
}
}
/// <summary>
/// A write got an answer from the Agent: <see cref="DriverHealth.LastSuccessfulRead"/> advances, and the state
/// follows the read / stream flags exactly as their own successes do (<see cref="PublishHealthy"/> — a write
/// cannot clear either path's degradation, but it is evidence the Agent is reachable).
/// </summary>
private void RecordWriteAnswered()
{
if (ReadHealth().State is DriverState.Healthy or DriverState.Degraded)
{
PublishHealthy(DateTime.UtcNow);
}
}
/// <summary>
/// Renders a value to the invariant string a PUT query carries, or reports that it has no such
/// form.
@@ -391,16 +391,39 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
/// </summary>
public Action? AfterWrite { get; set; }
/// <summary>Every Agent write REQUEST, in order: the device and the (id, value) pairs it carried.</summary>
public ConcurrentQueue<(string DeviceName, IReadOnlyList<KeyValuePair<string, string>> Pairs)> WriteRequests { get; } = new();
/// <summary>DataItem ids the Agent refuses: a request naming any of them is refused as a whole.</summary>
public HashSet<string> RefusedDataItems { get; } = new(StringComparer.Ordinal);
/// <inheritdoc/>
public Task<MTConnectWriteOutcome> WriteObservationAsync(
string deviceName, string dataItemId, string value, CancellationToken cancellationToken)
string deviceName, string dataItemId, string value, CancellationToken cancellationToken) =>
WriteObservationsAsync(deviceName, [new KeyValuePair<string, string>(dataItemId, value)], cancellationToken);
/// <inheritdoc/>
public Task<MTConnectWriteOutcome> WriteObservationsAsync(
string deviceName, IReadOnlyList<KeyValuePair<string, string>> observations, CancellationToken cancellationToken)
{
ObjectDisposedException.ThrowIf(IsDisposed, this);
cancellationToken.ThrowIfCancellationRequested();
Writes.Enqueue((deviceName, dataItemId, value));
WriteRequests.Enqueue((deviceName, observations));
foreach (var (dataItemId, value) in observations)
{
Writes.Enqueue((deviceName, dataItemId, value));
}
AfterWrite?.Invoke();
return WriteFailure is null ? Task.FromResult(WriteOutcome) : throw WriteFailure;
if (WriteFailure is not null)
{
throw WriteFailure;
}
return Task.FromResult(observations.Any(o => RefusedDataItems.Contains(o.Key))
? new MTConnectWriteOutcome(Accepted: false, AgentError: "UNSUPPORTED: refused data item")
: WriteOutcome);
}
/// <summary>
@@ -108,6 +108,23 @@ public sealed class MTConnectAgentClientWriteTests
.ShouldBe("http://stub-agent:5000/StubDevice?dev1_msg=hello");
}
/// <summary>R06:T-6 — several observations on one device are one POST carrying one escaped pair each.</summary>
[Fact]
public async Task Several_observations_are_one_request_with_one_pair_each()
{
var handler = new RecordingHandler(_ => Ok());
using var client = new MTConnectAgentClient(Options(), handler);
var outcome = await client.WriteObservationsAsync(
"StubDevice",
[new KeyValuePair<string, string>("dev1_msg", "a&b"), new KeyValuePair<string, string>("dev1_count", "7")],
CancellationToken.None);
outcome.Accepted.ShouldBeTrue();
handler.Requests.ShouldHaveSingleItem().Uri.AbsoluteUri
.ShouldBe("http://stub-agent:5000/StubDevice?dev1_msg=a%26b&dev1_count=7");
}
/// <summary>
/// A value carrying a space, an <c>&amp;</c> and a <c>=</c> — each of which silently ends the
/// query pair early if it reaches the wire raw — plus a DataItem id needing the same
@@ -0,0 +1,107 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
/// <summary>
/// Review 2026-10-21 R06:T-6 — writes were one POST per slot although the Agent's PUT surface accepts several
/// <c>id=value</c> pairs per device. Surviving slots now go out as one request per device, with a refused
/// multi-slot request retried slot by slot so one refused DataItem still costs only its own slot. And the health
/// contract: an Agent answer (accepted or refused) is the Agent talking, so it advances LastSuccessfulRead.
/// </summary>
public sealed class MTConnectBatchedWriteTests
{
private const string FixtureDeviceName = "VMC-3Axis";
private const string PosRaw = "Plant/MTConnect/dev1/Position";
private const string CountRaw = "Plant/MTConnect/dev1/PartCount";
private const string ProgramRaw = "Plant/MTConnect/dev1/Program";
private const uint Good = 0x00000000u;
private const uint BadCommunicationError = 0x80050000u;
private static CancellationToken Ct => TestContext.Current.CancellationToken;
private static RawTagEntry Entry(string rawPath, string dataItemId) =>
new(rawPath, $$"""{"fullName":"{{dataItemId}}"}""", WriteIdempotent: false, DeviceName: "dev1");
private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> WritableAsync()
{
var client = CannedAgentClient.FromFixtures();
var driver = new MTConnectDriver(
new MTConnectDriverOptions
{
AgentUri = "http://fixture-agent:5000",
Tags = [],
RawTags =
[
Entry(PosRaw, "dev1_pos"),
Entry(CountRaw, "dev1_partcount"),
Entry(ProgramRaw, "dev1_program"),
],
Write = new MTConnectWriteOptions { Enabled = true },
},
"mt-t6",
_ => client);
await driver.InitializeAsync("{}", Ct);
return (driver, client);
}
/// <summary>Three writes to three DataItems of one device are ONE Agent request carrying three pairs.</summary>
[Fact]
public async Task Writes_to_one_device_go_out_as_one_request()
{
var (driver, client) = await WritableAsync();
var results = await driver.WriteAsync(
[new WriteRequest(PosRaw, 1.5), new WriteRequest(CountRaw, 7), new WriteRequest(ProgramRaw, "O1000")], Ct);
results.ShouldAllBe(r => r.StatusCode == Good);
var request = client.WriteRequests.ShouldHaveSingleItem();
request.DeviceName.ShouldBe(FixtureDeviceName);
request.Pairs.Select(p => (p.Key, p.Value)).ShouldBe(
[("dev1_pos", "1.5"), ("dev1_partcount", "7"), ("dev1_program", "O1000")]);
}
/// <summary>Two writes to the SAME DataItem are never merged into one request — they land in request order.</summary>
[Fact]
public async Task Two_writes_to_one_data_item_stay_separate_and_ordered()
{
var (driver, client) = await WritableAsync();
await driver.WriteAsync([new WriteRequest(PosRaw, 1.0), new WriteRequest(CountRaw, 3), new WriteRequest(PosRaw, 2.0)], Ct);
client.WriteRequests.Count.ShouldBe(2);
client.Writes.Where(w => w.DataItemId == "dev1_pos").Select(w => w.Value).ShouldBe(["1", "2"]);
}
/// <summary>A refused multi-slot request is retried per slot: only the refused DataItem's slot fails.</summary>
[Fact]
public async Task A_refused_batch_is_retried_per_slot_so_only_the_refused_item_fails()
{
var (driver, client) = await WritableAsync();
client.RefusedDataItems.Add("dev1_partcount");
var results = await driver.WriteAsync(
[new WriteRequest(PosRaw, 1.5), new WriteRequest(CountRaw, 7), new WriteRequest(ProgramRaw, "O1000")], Ct);
results.Select(r => r.StatusCode).ShouldBe([Good, BadCommunicationError, Good]);
client.WriteRequests.Count.ShouldBe(4, "the batch, then one retry per slot");
}
/// <summary>A refusal is a reply: it advances LastSuccessfulRead (the Agent answered).</summary>
[Fact]
public async Task A_refused_write_is_an_answer_and_advances_the_stamp()
{
var (driver, client) = await WritableAsync();
client.RefusedDataItems.Add("dev1_pos");
var before = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
// The stamp is DateTime.UtcNow: wait (bounded) for the clock to tick past it so an advance is observable.
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
while (DateTime.UtcNow <= before && DateTime.UtcNow < deadline) await Task.Yield();
await driver.WriteAsync([new WriteRequest(PosRaw, 1.5)], Ct);
driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(before);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
}
}