rogee/wx-win-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stop data sync after account authorization revoke

Browse Source
This commit is contained in:
rogee
2026-09-22 13:02:55 +08:00
parent 51c9524437
commit e8a8b2f926
6 changed files with 159 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/validation/raw/collect-live-operations.sh
+20 -1
View File
@@ -125,6 +125,18 @@ after=$(sync_summary)
echo "{\"event\":\"agent-restart\",\"process\":$restart_process,\"before\":$before,\"after\":$after}"
run_ps "$stop_ps" >/dev/null 2>&1
# Create a real, valid local pending batch while the control plane is unavailable.
# The agent must later discard this queued body after authorization is revoked.
stop_server
run_ps "Remove-Item \"$WINDOWS_DIR/data/remote-data-sync-state.json\",\"$WINDOWS_DIR/data/remote-data-queue.json\" -Force -ErrorAction SilentlyContinue" >/dev/null 2>&1
pending_process=$(start_tray)
pending_queue=$(queue_json)
run_ps "$stop_ps" >/dev/null 2>&1
start_server
for _ in $(seq 1 20); do
sleep 1
curl -fsS http://127.0.0.1:8090/healthz >/dev/null 2>&1 && break
done
web_token=$(curl -fsS -X POST http://127.0.0.1:8090/v1/auth/login -H 'Content-Type: application/json' -d "{\"username\":\"admin\",\"password\":\"$WEB_PASSWORD\"}" | jq -r .access_token)
revoke=$(curl -sS -o "$root/revoke" -w '%{http_code}' -X POST -H "Authorization: Bearer $web_token" "http://127.0.0.1:8090/v1/data/accounts/$ACCOUNT_ID/revoke")
conv=$(curl -sS -o "$root/conv" -w '%{http_code}' -H "Authorization: Bearer $web_token" "http://127.0.0.1:8090/v1/data/accounts/$ACCOUNT_ID/conversations")
@@ -132,7 +144,14 @@ msgs=$(curl -sS -o "$root/msg" -w '%{http_code}' -H "Authorization: Bearer $web_
sync=$(curl -sS -o "$root/sync" -w '%{http_code}' -H "Authorization: Bearer $web_token" "http://127.0.0.1:8090/v1/data/accounts/$ACCOUNT_ID/sync-status")
reject_payload=$(jq -nc --arg account "$ACCOUNT_ID" '{node_id:"local-node",account_id:$account,batch_id:"revoked-batch-harness",source_generation:$account,stream_key:"messages",sequence:5,cursor_start:"4",cursor_end:"5",payload_hash:"revoked-batch-hash",coverage_state:"complete",conversations:[],messages:[{message_id:"revoked-message-harness",chat_id:"filehelper",chat_type:"Private",source_message_id:"revoked-source-harness",direction:"incoming",message_type:"text",text:"fixture-revoked",source_time:"2026-09-22T03:00:00Z",observed_at:"2026-09-22T03:00:00Z",source_version:"harness",payload_hash:"revoked-message-hash"}]}')
batch_rejected=$(curl -sS -o "$root/rejected" -w '%{http_code}' -X POST -H "Authorization: Bearer $NODE_TOKEN" -H 'Content-Type: application/json' --data-binary "$reject_payload" http://127.0.0.1:8090/v1/data/batches)
echo "{\"event\":\"authorization-revoke\",\"revoke\":$revoke,\"conversations\":$conv,\"messages\":$msgs,\"sync_status\":$sync,\"batch_rejected\":$batch_rejected}"
# Start the agent only after revocation. Its reconciliation/flush path must drop the
# pre-existing queue and keep it empty instead of collecting or retransmitting.
revoked_process=$(start_tray)
revoked_queue=$(queue_json)
sleep 12
revoked_queue_after=$(queue_json)
echo "{\"event\":\"authorization-revoke\",\"pending_process\":$pending_process,\"pending_queue\":$pending_queue,\"revoke\":$revoke,\"conversations\":$conv,\"messages\":$msgs,\"sync_status\":$sync,\"batch_rejected\":$batch_rejected,\"revoked_process\":$revoked_process,\"revoked_queue\":$revoked_queue,\"revoked_queue_after\":$revoked_queue_after}"
run_ps "$stop_ps" >/dev/null 2>&1
restore_process=$(start_tray)
sleep 5
web_token=$(curl -fsS -X POST http://127.0.0.1:8090/v1/auth/login -H 'Content-Type: application/json' -d "{\"username\":\"admin\",\"password\":\"$WEB_PASSWORD\"}" | jq -r .access_token)
docs/validation/raw/verify-p4-live-operations.sh
+1 -1
View File
@@ -17,7 +17,7 @@ jq -s -e '
($e["offline-queue"].queue.queue_pending > 0 and $e["offline-queue"].queue.queue_bytes > 0 and $e["offline-queue"].process.session == 1) and
($e["replay"].queue.queue_pending == 0 and $e["replay"].sync.state == "complete" and $e["replay"].shard.coverage == "complete" and $e["replay"].shard.integrity == "ok") and
($e["agent-restart"].process.session == 1 and $e["agent-restart"].before.confirmed_sequence == $e["agent-restart"].after.confirmed_sequence and $e["agent-restart"].after.state == "complete") and
($e["authorization-revoke"].revoke == 200 and $e["authorization-revoke"].conversations == 403 and $e["authorization-revoke"].messages == 403 and $e["authorization-revoke"].sync_status == 200 and $e["authorization-revoke"].batch_rejected == 403) and
($e["authorization-revoke"].pending_process.session == 1 and $e["authorization-revoke"].pending_queue.queue_pending > 0 and $e["authorization-revoke"].pending_queue.queue_bytes > 0 and $e["authorization-revoke"].revoke == 200 and $e["authorization-revoke"].conversations == 403 and $e["authorization-revoke"].messages == 403 and $e["authorization-revoke"].sync_status == 200 and $e["authorization-revoke"].batch_rejected == 403 and $e["authorization-revoke"].revoked_process.session == 1 and $e["authorization-revoke"].revoked_queue.queue_pending == 0 and $e["authorization-revoke"].revoked_queue_after.queue_pending == 0) and
($e["authorization-restore"].process.session == 1 and $e["authorization-restore"].conversations == 200 and $e["authorization-restore"].messages == 200 and $e["authorization-restore"].queue.queue_pending == 0) and
($e["control-plane-restart"].before.integrity == "ok" and $e["control-plane-restart"].after.integrity == "ok" and $e["control-plane-restart"].before.messages == $e["control-plane-restart"].after.messages and $e["control-plane-restart"].before.batches == $e["control-plane-restart"].after.batches)
' "$file" >/dev/null
node-agent/WxAgent.Core/RemoteControlClient.cs
+21 -3
View File
@@ -190,24 +190,42 @@ public sealed class RemoteControlClient : IDisposable
cancellationToken, RemoteDataProtocol.MaxBatchBytes).ConfigureAwait(false);
}
public Task<IReadOnlyList<RemoteSyncBatch>> FlushDataBatchesAsync(
RemoteDataBatchQueue queue,
ReportingConfig reportingConfig,
CancellationToken cancellationToken = default) =>
FlushDataBatchesAsync(queue, reportingConfig, new HashSet<string>(StringComparer.Ordinal), cancellationToken);
public async Task<IReadOnlyList<RemoteSyncBatch>> FlushDataBatchesAsync(
RemoteDataBatchQueue queue,
ReportingConfig reportingConfig,
ISet<string> blockedAccountIds,
CancellationToken cancellationToken = default)
{
EnsureAuthenticated();
var confirmed = new List<RemoteSyncBatch>();
foreach (var batch in queue.Pending())
{
if (blockedAccountIds.Contains(batch.AccountId))
continue;
var filtered = RemoteDataBatchAuthorization.Filter(reportingConfig, batch, out _);
if (filtered is null)
{
queue.Drop(batch.BatchId);
continue;
}
var acknowledgement = await SubmitDataBatchAsync(reportingConfig, filtered, cancellationToken).ConfigureAwait(false);
if (acknowledgement.Accepted && queue.MarkConfirmed(acknowledgement))
confirmed.Add(filtered);
try
{
var acknowledgement = await SubmitDataBatchAsync(reportingConfig, filtered, cancellationToken).ConfigureAwait(false);
if (acknowledgement.Accepted && queue.MarkConfirmed(acknowledgement))
confirmed.Add(filtered);
}
catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized")
{
// Authorization revocation is terminal for locally buffered content: do not retain or retry it.
queue.DropAllForAccount(batch.AccountId);
blockedAccountIds.Add(batch.AccountId);
}
}
return confirmed;
}
node-agent/WxAgent.Core/RemoteDataSync.cs
+11
View File
@@ -362,6 +362,17 @@ public sealed class RemoteDataBatchQueue
}
}
public int DropAllForAccount(string accountId)
{
lock (gate)
{
var removed = state.Items.RemoveAll(item => string.Equals(item.AccountId, accountId, StringComparison.Ordinal));
if (removed > 0)
SaveLocked();
return removed;
}
}
public bool MarkConfirmed(RemoteDataBatchAck acknowledgement)
{
lock (gate)
node-agent/WxAgent.Service/RemoteAgentHostedService.cs
+29 -7
View File
@@ -48,6 +48,7 @@ public sealed class RemoteAgentHostedService(
var syncState = options.EnableDataSync
? new RemoteDataSyncStateStore(Path.Combine(options.DataDirectory, "remote-data-sync-state.json"))
: null;
var blockedDataSyncAccounts = new HashSet<string>(StringComparer.Ordinal);
var lastDataSyncAt = DateTimeOffset.MinValue;
string? registeredActiveAccountId = null;
bool? registeredActiveAccountVerified = null;
@@ -78,6 +79,7 @@ public sealed class RemoteAgentHostedService(
registeredActiveAccountId = snapshot.ActiveAccountId;
registeredActiveAccountVerified = snapshot.ActiveAccountVerified;
registeredReportingConfigVersion = reporting.ConfigVersion;
blockedDataSyncAccounts.Clear();
retry = RetryDelay;
}
@@ -87,20 +89,24 @@ public sealed class RemoteAgentHostedService(
if (dataQueue is not null && syncState is not null && dataCollector is not null
&& DateTimeOffset.UtcNow - lastDataSyncAt >= TimeSpan.FromSeconds(options.DataSyncIntervalSeconds))
{
await ReconcileDataStatusAsync(client, reporting, dataQueue, syncState, stoppingToken);
await ReconcileDataStatusAsync(client, reporting, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
var blockedBeforeFlush = blockedDataSyncAccounts.ToHashSet(StringComparer.Ordinal);
try
{
var confirmedBatches = await client.FlushDataBatchesAsync(dataQueue, reporting, stoppingToken);
var confirmedBatches = await client.FlushDataBatchesAsync(dataQueue, reporting, blockedDataSyncAccounts, stoppingToken);
foreach (var confirmedBatch in confirmedBatches)
syncState.MarkConfirmed(confirmedBatch);
foreach (var accountId in blockedDataSyncAccounts.Except(blockedBeforeFlush, StringComparer.Ordinal))
logger.LogWarning("Data sync authorization was revoked; accountId={AccountId}; pending content was discarded and collection is paused.", accountId);
}
catch (Exception exception) when (exception is HttpRequestException or RemoteClientException)
{
logger.LogWarning("Data sync delivery deferred; type={ExceptionType}.", exception.GetType().Name);
}
// Collection must continue while the platform is unreachable; the durable queue is
// the offline buffer and will be flushed on the next successful connection.
await CollectDataBatchesAsync(remote, reporting, dataCollector, dataQueue, syncState, stoppingToken);
// the offline buffer and will be flushed on the next successful connection. Revoked
// accounts are the exception: their pending content is discarded and collection stops.
await CollectDataBatchesAsync(remote, reporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
lastDataSyncAt = DateTimeOffset.UtcNow;
}
var pollAccountIds = reporting.Accounts
@@ -133,14 +139,14 @@ public sealed class RemoteAgentHostedService(
{
logger.LogWarning("Remote control-plane request failed; code={Code}; status={StatusCode}; correlationId={CorrelationId}.",
exception.Code, exception.StatusCode, exception.CorrelationId);
await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, stoppingToken);
await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
await DelayAsync(retry, stoppingToken);
retry = TimeSpan.FromSeconds(Math.Min(retry.TotalSeconds * 2, 30));
}
catch (Exception exception)
{
logger.LogWarning("Remote agent cycle failed; type={ExceptionType}.", exception.GetType().Name);
await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, stoppingToken);
await TryCollectOfflineDataBatchesAsync(remote, activeReporting, dataCollector, dataQueue, syncState, blockedDataSyncAccounts, stoppingToken);
await DelayAsync(retry, stoppingToken);
retry = TimeSpan.FromSeconds(Math.Min(retry.TotalSeconds * 2, 30));
}
@@ -153,11 +159,14 @@ public sealed class RemoteAgentHostedService(
IRemoteDataCollector collector,
RemoteDataBatchQueue queue,
RemoteDataSyncStateStore state,
ISet<string> blockedAccountIds,
CancellationToken cancellationToken)
{
foreach (var account in reporting.Accounts.Where(item => item.Enabled))
{
cancellationToken.ThrowIfCancellationRequested();
if (blockedAccountIds.Contains(account.AccountId))
continue;
var scopes = AuthorizedChatScopes(reporting, account.AccountId);
if (scopes.Count == 0 || queue.Pending().Any(batch => string.Equals(batch.AccountId, account.AccountId, StringComparison.Ordinal)
&& string.Equals(batch.StreamKey, "messages", StringComparison.Ordinal)))
@@ -191,13 +200,14 @@ public sealed class RemoteAgentHostedService(
IRemoteDataCollector? collector,
RemoteDataBatchQueue? queue,
RemoteDataSyncStateStore? state,
ISet<string> blockedAccountIds,
CancellationToken cancellationToken)
{
if (reporting is null || collector is null || queue is null || state is null)
return;
try
{
await CollectDataBatchesAsync(remote, reporting, collector, queue, state, cancellationToken).ConfigureAwait(false);
await CollectDataBatchesAsync(remote, reporting, collector, queue, state, blockedAccountIds, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
@@ -214,6 +224,7 @@ public sealed class RemoteAgentHostedService(
ReportingConfig reporting,
RemoteDataBatchQueue queue,
RemoteDataSyncStateStore state,
ISet<string> blockedAccountIds,
CancellationToken cancellationToken)
{
foreach (var account in reporting.Accounts.Where(item => item.Enabled))
@@ -229,12 +240,23 @@ public sealed class RemoteAgentHostedService(
// Older control planes do not expose the reconciliation endpoint; retain the legacy ACK path.
continue;
}
catch (RemoteClientException exception) when (exception.StatusCode == 403 && exception.Code == "AccountNotAuthorized")
{
var droppedOnRevoke = queue.DropAllForAccount(account.AccountId);
blockedAccountIds.Add(account.AccountId);
logger.LogWarning(
"Data sync authorization is revoked; accountId={AccountId}; pending content was discarded and collection is paused; droppedPendingBatches={DroppedPendingBatches}.",
account.AccountId, droppedOnRevoke);
continue;
}
catch (HttpRequestException)
{
// A disconnected platform must not prevent local DB collection.
return;
}
if (blockedAccountIds.Remove(account.AccountId))
logger.LogInformation("Data sync authorization was restored; accountId={AccountId}; collection is resumed.", account.AccountId);
if (!state.Reconcile(remote, account.AccountId, "messages"))
continue;
var dropped = queue.DropForAccount(account.AccountId, "messages");
tests/node-agent/WxAgent.Core.Tests/RemoteControlTests.cs
+77
View File
@@ -335,6 +335,51 @@ public sealed class RemoteControlClientTests
Assert.All(handler.Requests, request => Assert.Equal("Bearer node-token", request.Authorization));
}
[Fact]
public async Task FlushDropsAllQueuedBatchesAndBlocksAccountAfterRevocation()
{
var directory = Directory.CreateTempSubdirectory("wxagent-revoke-");
try
{
var config = new ReportingConfig
{
Enabled = true,
ConfigVersion = 1,
Accounts = [new AccountReportingConfig
{
AccountId = "account-a", Enabled = true,
AllowedChats = [new AllowedChat { Type = ReportingChatType.Private, ChatId = "chat-a", Enabled = true, IdentityVerified = true }]
}]
};
var queue = new RemoteDataBatchQueue(Path.Combine(directory.FullName, "queue.json"));
Assert.True(queue.Enqueue(config, SyncBatch("batch-1")).Accepted);
Assert.True(queue.Enqueue(config, SyncBatch("batch-2")).Accepted);
var handler = new RevokedBatchHandler();
using var http = new HttpClient(handler);
using var client = new RemoteControlClient(new RemoteAgentOptions
{
AuthAddress = "http://127.0.0.1:8090",
Token = "node-token",
NodeId = "node-1"
}, http);
await client.RegisterAsync(new RemoteNodeRegistration("node-1", "test", RemoteProtocol.Version, ["db-messages"], 1, []));
var blockedAccounts = new HashSet<string>(StringComparer.Ordinal);
var confirmed = await client.FlushDataBatchesAsync(queue, config, blockedAccounts);
Assert.Empty(confirmed);
Assert.Contains("account-a", blockedAccounts);
Assert.Empty(queue.Pending());
Assert.Empty(new RemoteDataBatchQueue(Path.Combine(directory.FullName, "queue.json")).Pending());
Assert.Equal(1, handler.BatchRequests);
}
finally
{
directory.Delete(true);
}
}
[Fact]
public async Task MissingRemoteCredentialsAreRejectedWithoutNetworkAccess()
{
@@ -345,6 +390,38 @@ public sealed class RemoteControlClientTests
Assert.Empty(handler.Requests);
}
private static RemoteSyncBatch SyncBatch(string batchId) => new(
"node-1", "account-a", batchId, "generation-a", "messages", 1, "{}", "{\"chat-a\\u001fmessage/a.db\":1}",
"hash-" + batchId, "complete", [],
[new RemoteSyncMessage(batchId, "chat-a", ReportingChatType.Private, batchId, "incoming", "text", "queued secret",
DateTimeOffset.UtcNow, DateTimeOffset.UtcNow, "wx-1", "payload-" + batchId)]);
private sealed class RevokedBatchHandler : HttpMessageHandler
{
public int BatchRequests { get; private set; }
protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
if (request.RequestUri!.AbsolutePath == "/v1/data/batches")
{
BatchRequests++;
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.Forbidden)
{
Content = new StringContent("{\"error\":{\"code\":\"AccountNotAuthorized\",\"message\":\"revoked\"}}", Encoding.UTF8, "application/json")
});
}
if (request.RequestUri.AbsolutePath == "/v1/nodes/register")
{
var body = JsonSerializer.Serialize(new RemoteNodeRegistrationResponse("node-1", RemoteNodeStatus.Online, true, "register"), RemoteJson.Options);
return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StringContent(body, Encoding.UTF8, "application/json")
});
}
throw new InvalidOperationException("Unexpected request " + request.RequestUri.AbsolutePath);
}
}
private sealed class RecordingHandler : HttpMessageHandler
{
public List<RecordedRequest> Requests { get; } = [];