Implement worker advise commands

2026-04-26 18:41:10 -04:00
parent 0fd954d94c
commit a871f2f2e5
11 changed files with 734 additions and 1 deletions
@@ -177,6 +177,30 @@ public sealed class MxAccessCommandExecutorTests
        Assert.Empty(await session.GetRegisteredItemHandlesAsync());
    }

+    [Fact]
+    public async Task DispatchAsync_RemoveItemWithAdvisedHandle_RemovesTrackedAdviceAfterMxAccessSucceeds()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 148,
+            addItemHandle: 603);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-advised-remove", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-advised-remove", 148, "Galaxy.Tag.Value"));
+        await session.DispatchAsync(CreateAdviseCommand("advise-before-remove", 148, 603));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateRemoveItemCommand(
+            "remove-advised-item",
+            148,
+            603));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Empty(await session.GetRegisteredItemHandlesAsync());
+        Assert.Empty(await session.GetRegisteredAdviceHandlesAsync());
+    }
+
    [Fact]
    public async Task DispatchAsync_RemoveItemWithCrossServerHandle_PreservesHResultAndKeepsTrackedItemHandle()
    {
@@ -238,6 +262,158 @@ public sealed class MxAccessCommandExecutorTests
        Assert.Empty(await session.GetRegisteredItemHandlesAsync());
    }

+    [Fact]
+    public async Task DispatchAsync_Advise_CallsMxAccessOnStaAndTracksPlainAdvice()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 52,
+            addItemHandle: 505);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-advise", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-advise", 52, "Galaxy.Tag.Value"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateAdviseCommand(
+            "advise",
+            52,
+            505));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(0, reply.Hresult);
+        Assert.Equal(52, fakeComObject.AdviseServerHandle);
+        Assert.Equal(505, fakeComObject.AdvisedItemHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.AdviseThreadId);
+
+        RegisteredAdviceHandle adviceHandle = Assert.Single(
+            await session.GetRegisteredAdviceHandlesAsync());
+        Assert.Equal(52, adviceHandle.ServerHandle);
+        Assert.Equal(505, adviceHandle.ItemHandle);
+        Assert.Equal(MxAccessAdviceKind.Plain, adviceHandle.AdviceKind);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_AdviseSupervisory_CallsDistinctMxAccessMethodAndTracksSupervisoryAdvice()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 53,
+            addItemHandle: 506);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-supervisory", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-supervisory", 53, "Galaxy.Tag.Value"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateAdviseSupervisoryCommand(
+            "advise-supervisory",
+            53,
+            506));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(53, fakeComObject.AdviseSupervisoryServerHandle);
+        Assert.Equal(506, fakeComObject.AdviseSupervisoryItemHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.AdviseSupervisoryThreadId);
+        Assert.Null(fakeComObject.AdviseServerHandle);
+
+        RegisteredAdviceHandle adviceHandle = Assert.Single(
+            await session.GetRegisteredAdviceHandlesAsync());
+        Assert.Equal(53, adviceHandle.ServerHandle);
+        Assert.Equal(506, adviceHandle.ItemHandle);
+        Assert.Equal(MxAccessAdviceKind.Supervisory, adviceHandle.AdviceKind);
+    }
+
+    [Fact]
+    public async Task DispatchAsync_UnAdvise_CallsMxAccessOnStaAndRemovesTrackedAdvice()
+    {
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 54,
+            addItemHandle: 507);
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-unadvise", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-unadvise", 54, "Galaxy.Tag.Value"));
+        await session.DispatchAsync(CreateAdviseCommand("advise-before-unadvise", 54, 507));
+        await session.DispatchAsync(CreateAdviseSupervisoryCommand("supervisory-before-unadvise", 54, 507));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateUnAdviseCommand(
+            "unadvise",
+            54,
+            507));
+
+        Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code);
+        Assert.Equal(54, fakeComObject.UnAdviseServerHandle);
+        Assert.Equal(507, fakeComObject.UnAdvisedItemHandle);
+        Assert.Equal(runtime.StaThreadId, fakeComObject.UnAdviseThreadId);
+        Assert.Empty(await session.GetRegisteredAdviceHandlesAsync());
+    }
+
+    [Fact]
+    public async Task DispatchAsync_AdviseWhenMxAccessThrows_PreservesHResultAndDoesNotTrackAdvice()
+    {
+        const int hresult = unchecked((int)0x80070057);
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 55,
+            addItemHandle: 508,
+            adviseException: new COMException("Invalid item handle.", hresult));
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-advise-failure", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-advise-failure", 55, "Galaxy.Tag.Value"));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateAdviseCommand(
+            "advise-failure",
+            55,
+            999));
+
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(hresult, reply.Hresult);
+        Assert.Contains("0x80070057", reply.DiagnosticMessage);
+        Assert.Equal(55, fakeComObject.AdviseServerHandle);
+        Assert.Equal(999, fakeComObject.AdvisedItemHandle);
+        Assert.Empty(await session.GetRegisteredAdviceHandlesAsync());
+    }
+
+    [Fact]
+    public async Task DispatchAsync_UnAdviseWhenMxAccessThrows_PreservesHResultAndKeepsTrackedAdvice()
+    {
+        const int hresult = unchecked((int)0x80070057);
+        FakeMxAccessComObject fakeComObject = new(
+            registerHandle: 56,
+            addItemHandle: 509,
+            unAdviseException: new COMException("Invalid item handle.", hresult));
+        FakeMxAccessComObjectFactory factory = new(fakeComObject);
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+        await session.DispatchAsync(CreateRegisterCommand("register-before-unadvise-failure", "client-a"));
+        await session.DispatchAsync(CreateAddItemCommand("add-before-unadvise-failure", 56, "Galaxy.Tag.Value"));
+        await session.DispatchAsync(CreateAdviseCommand("advise-before-unadvise-failure", 56, 509));
+
+        MxCommandReply reply = await session.DispatchAsync(CreateUnAdviseCommand(
+            "unadvise-failure",
+            56,
+            509));
+
+        Assert.Equal(ProtocolStatusCode.MxaccessFailure, reply.ProtocolStatus.Code);
+        Assert.True(reply.HasHresult);
+        Assert.Equal(hresult, reply.Hresult);
+        Assert.Contains("0x80070057", reply.DiagnosticMessage);
+        Assert.Equal(56, fakeComObject.UnAdviseServerHandle);
+        Assert.Equal(509, fakeComObject.UnAdvisedItemHandle);
+
+        RegisteredAdviceHandle adviceHandle = Assert.Single(
+            await session.GetRegisteredAdviceHandlesAsync());
+        Assert.Equal(MxAccessAdviceKind.Plain, adviceHandle.AdviceKind);
+    }
+
    [Fact]
    public async Task DispatchAsync_RegisterWithoutPayload_ReturnsInvalidRequest()
    {
@@ -278,6 +454,26 @@ public sealed class MxAccessCommandExecutorTests
        Assert.Null(factory.FakeComObject.AddItemDefinition);
    }

+    [Fact]
+    public async Task DispatchAsync_AdviseWithoutPayload_ReturnsInvalidRequest()
+    {
+        FakeMxAccessComObjectFactory factory = new(new FakeMxAccessComObject(registerHandle: 57));
+        using StaRuntime runtime = CreateRuntime();
+        using MxAccessStaSession session = new(runtime, factory, new NoopEventSink());
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply reply = await session.DispatchAsync(new StaCommand(
+            "session-1",
+            "missing-advise-payload",
+            new MxCommand
+            {
+                Kind = MxCommandKind.Advise,
+            }));
+
+        Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code);
+        Assert.Null(factory.FakeComObject.AdviseServerHandle);
+    }
+
    private static StaCommand CreateRegisterCommand(
        string correlationId,
        string clientName)
@@ -371,6 +567,63 @@ public sealed class MxAccessCommandExecutorTests
            });
    }

+    private static StaCommand CreateAdviseCommand(
+        string correlationId,
+        int serverHandle,
+        int itemHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.Advise,
+                Advise = new AdviseCommand
+                {
+                    ServerHandle = serverHandle,
+                    ItemHandle = itemHandle,
+                },
+            });
+    }
+
+    private static StaCommand CreateUnAdviseCommand(
+        string correlationId,
+        int serverHandle,
+        int itemHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.UnAdvise,
+                UnAdvise = new UnAdviseCommand
+                {
+                    ServerHandle = serverHandle,
+                    ItemHandle = itemHandle,
+                },
+            });
+    }
+
+    private static StaCommand CreateAdviseSupervisoryCommand(
+        string correlationId,
+        int serverHandle,
+        int itemHandle)
+    {
+        return new StaCommand(
+            "session-1",
+            correlationId,
+            new MxCommand
+            {
+                Kind = MxCommandKind.AdviseSupervisory,
+                AdviseSupervisory = new AdviseSupervisoryCommand
+                {
+                    ServerHandle = serverHandle,
+                    ItemHandle = itemHandle,
+                },
+            });
+    }
+
    private static StaRuntime CreateRuntime()
    {
        return new StaRuntime(
@@ -388,6 +641,9 @@ public sealed class MxAccessCommandExecutorTests
        private readonly Exception? addItemException;
        private readonly Exception? addItem2Exception;
        private readonly Exception? removeItemException;
+        private readonly Exception? adviseException;
+        private readonly Exception? unAdviseException;
+        private readonly Exception? adviseSupervisoryException;

        public FakeMxAccessComObject(
            int registerHandle,
@@ -396,7 +652,10 @@ public sealed class MxAccessCommandExecutorTests
            Exception? unregisterException = null,
            Exception? addItemException = null,
            Exception? addItem2Exception = null,
-            Exception? removeItemException = null)
+            Exception? removeItemException = null,
+            Exception? adviseException = null,
+            Exception? unAdviseException = null,
+            Exception? adviseSupervisoryException = null)
        {
            this.registerHandle = registerHandle;
            this.addItemHandle = addItemHandle;
@@ -405,6 +664,9 @@ public sealed class MxAccessCommandExecutorTests
            this.addItemException = addItemException;
            this.addItem2Exception = addItem2Exception;
            this.removeItemException = removeItemException;
+            this.adviseException = adviseException;
+            this.unAdviseException = unAdviseException;
+            this.adviseSupervisoryException = adviseSupervisoryException;
        }

        public string? RegisteredClientName { get; private set; }
@@ -435,6 +697,24 @@ public sealed class MxAccessCommandExecutorTests

        public int? RemoveItemThreadId { get; private set; }

+        public int? AdviseServerHandle { get; private set; }
+
+        public int? AdvisedItemHandle { get; private set; }
+
+        public int? AdviseThreadId { get; private set; }
+
+        public int? UnAdviseServerHandle { get; private set; }
+
+        public int? UnAdvisedItemHandle { get; private set; }
+
+        public int? UnAdviseThreadId { get; private set; }
+
+        public int? AdviseSupervisoryServerHandle { get; private set; }
+
+        public int? AdviseSupervisoryItemHandle { get; private set; }
+
+        public int? AdviseSupervisoryThreadId { get; private set; }
+
        public int Register(string clientName)
        {
            RegisteredClientName = clientName;
@@ -501,6 +781,48 @@ public sealed class MxAccessCommandExecutorTests
                throw removeItemException;
            }
        }
+
+        public void Advise(
+            int serverHandle,
+            int itemHandle)
+        {
+            AdviseServerHandle = serverHandle;
+            AdvisedItemHandle = itemHandle;
+            AdviseThreadId = Environment.CurrentManagedThreadId;
+
+            if (adviseException is not null)
+            {
+                throw adviseException;
+            }
+        }
+
+        public void UnAdvise(
+            int serverHandle,
+            int itemHandle)
+        {
+            UnAdviseServerHandle = serverHandle;
+            UnAdvisedItemHandle = itemHandle;
+            UnAdviseThreadId = Environment.CurrentManagedThreadId;
+
+            if (unAdviseException is not null)
+            {
+                throw unAdviseException;
+            }
+        }
+
+        public void AdviseSupervisory(
+            int serverHandle,
+            int itemHandle)
+        {
+            AdviseSupervisoryServerHandle = serverHandle;
+            AdviseSupervisoryItemHandle = itemHandle;
+            AdviseSupervisoryThreadId = Environment.CurrentManagedThreadId;
+
+            if (adviseSupervisoryException is not null)
+            {
+                throw adviseSupervisoryException;
+            }
+        }
    }

    private sealed class FakeMxAccessComObjectFactory : IMxAccessComObjectFactory
@@ -215,6 +215,127 @@ public sealed class MxAccessLiveComCreationTests
        }
    }

+    [Fact]
+    public async Task AdviseAndUnAdvise_WhenOptedIn_RoundTripsInstalledMxAccessSubscription()
+    {
+        if (!RunLiveMxAccessTests())
+        {
+            return;
+        }
+
+        using MxAccessStaSession session = new();
+        await session.StartAsync(workerProcessId: 1234);
+
+        MxCommandReply registerReply = await RegisterLiveSessionAsync(session, "live-advise-register");
+        int serverHandle = registerReply.Register.ServerHandle;
+        int itemHandle = 0;
+        bool advised = false;
+
+        try
+        {
+            MxCommandReply addItemReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-advise-add-item",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.AddItem,
+                    AddItem = new AddItemCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemDefinition = GetLiveAddItemReference(),
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code);
+            Assert.True(addItemReply.AddItem.ItemHandle > 0);
+            itemHandle = addItemReply.AddItem.ItemHandle;
+
+            MxCommandReply adviseReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-advise",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.Advise,
+                    Advise = new AdviseCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemHandle = itemHandle,
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code);
+            advised = true;
+
+            MxCommandReply unAdviseReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-unadvise",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.UnAdvise,
+                    UnAdvise = new UnAdviseCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemHandle = itemHandle,
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, unAdviseReply.ProtocolStatus.Code);
+            advised = false;
+
+            MxCommandReply removeItemReply = await session.DispatchAsync(new StaCommand(
+                "session-1",
+                "live-advise-remove-item",
+                new MxCommand
+                {
+                    Kind = MxCommandKind.RemoveItem,
+                    RemoveItem = new RemoveItemCommand
+                    {
+                        ServerHandle = serverHandle,
+                        ItemHandle = itemHandle,
+                    },
+                }));
+
+            Assert.Equal(ProtocolStatusCode.Ok, removeItemReply.ProtocolStatus.Code);
+            itemHandle = 0;
+        }
+        finally
+        {
+            if (advised && itemHandle > 0)
+            {
+                await session.DispatchAsync(new StaCommand(
+                    "session-1",
+                    "live-unadvise-cleanup",
+                    new MxCommand
+                    {
+                        Kind = MxCommandKind.UnAdvise,
+                        UnAdvise = new UnAdviseCommand
+                        {
+                            ServerHandle = serverHandle,
+                            ItemHandle = itemHandle,
+                        },
+                    }));
+            }
+
+            if (itemHandle > 0)
+            {
+                await session.DispatchAsync(new StaCommand(
+                    "session-1",
+                    "live-advise-remove-item-cleanup",
+                    new MxCommand
+                    {
+                        Kind = MxCommandKind.RemoveItem,
+                        RemoveItem = new RemoveItemCommand
+                        {
+                            ServerHandle = serverHandle,
+                            ItemHandle = itemHandle,
+                        },
+                    }));
+            }
+
+            await UnregisterLiveSessionAsync(session, serverHandle, "live-advise-unregister");
+        }
+    }
+
    private static bool RunLiveMxAccessTests()
    {
        return string.Equals(