@samitouri / QOSAMI-WSL-2 / commits / 75cd0f04

Wslc events (#40971)

Kevin Vega committed Sep 3, 2026 at 18:06 UTC 75cd0f046957bbd9ffeb4e862621f8801dcb2651

18 files changed +1008 -91

doc/docs/api-reference/c/error-codes.md

@@ -18,6 +18,8 @@

18 #define WSLC_E_VOLUME_NOT_AVAILABLE MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 14) /* 0x8004060E */

19 #define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 15) /* 0x8004060F */

20 #define WSLC_E_VM_NOT_RUNNING MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */

21 +#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */

22 +#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 18) /* 0x80040612 */

23 ```

25 | Symbol | Hex Value |

@@ -39,5 +41,7 @@

41 | `WSLC_E_VOLUME_NOT_AVAILABLE` | `0x8004060E` |

42 | `WSLC_E_SESSION_NOT_FOUND` | `0x8004060F` |

43 | `WSLC_E_VM_NOT_RUNNING` | `0x80040610` |

44 +| `WSLC_E_EVENTS_LOST` | `0x80040611` |

45 +| `WSLC_E_EVENT_STREAM_FINISHED` | `0x80040612` |

47 ---

localization/strings/en-US/Resources.resw

@@ -2371,6 +2371,10 @@ For privacy information about this product please visit https://aka.ms/privacy.<

2371 <value>Invalid name: '{}'</value>

2372 <comment>{FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated</comment>

2373 </data>

2374 + <data name = "MessageWslcEventsInvalidTimeWindow" xml:space = "preserve" >

2375 + <value>`since` time ({}) cannot be after `until` time ({})</value>

2376 + <comment>{FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated{Locked="`since`"}{Locked="`until`"}</comment>

2377 + </data>

2378 <data name = "MessagePathNotAbsolute" xml:space = "preserve" >

2379 <value>Path is not absolute: '{}'</value>

2380 <comment>{FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated</comment>

msipackage/package.wix.in

@@ -359,6 +359,14 @@

359 </RegistryKey>

360 </RegistryKey>

361

362 +

363 + <RegistryKey Root="HKCR" Key="Interface\{7EC66D3B-D098-4D48-B69E-69166F6C4745}">

364 + <RegistryValue Value="IWSLCEventStream" Type="string" />

365 + <RegistryKey Key="ProxyStubClsid32">

366 + <RegistryValue Value="{4EA0C6DD-E9FF-48E7-994E-13A31D10DC60}" Type="string" />

367 + </RegistryKey>

368 + </RegistryKey>

369 +

370

371 <RegistryKey Root="HKCR" Key="Interface\{8C5A7B14-9D26-4FAE-AB31-7E5BC23F4801}">

372 <RegistryValue Value="ICrashDumpCallback" Type="string" />

src/windows/WslcSDK/wslcsdk.h

@@ -44,6 +44,8 @@ EXTERN_C_START

44 #define WSLC_E_VOLUME_NOT_AVAILABLE MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 14) /* 0x8004060E */

45 #define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 15) /* 0x8004060F */

46 #define WSLC_E_VM_NOT_RUNNING MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */

47 +#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */

48 +#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 18) /* 0x80040612 */

50 // Session values

51 #define WSLC_SESSION_OPTIONS_SIZE 72

src/windows/common/wslutil.cpp

@@ -180,6 +180,8 @@ static const std::map<HRESULT, LPCWSTR> g_commonErrors{

180 X(WSLC_E_NETWORK_NOT_FOUND),

181 X(WSLC_E_SESSION_NOT_FOUND),

182 X(WSLC_E_VM_NOT_RUNNING),

183 + X(WSLC_E_EVENTS_LOST),

184 + X(WSLC_E_EVENT_STREAM_FINISHED),

185 X(WSLC_E_WU_SEARCH_FAILED),

186 X_WIN32(RPC_S_SERVER_UNAVAILABLE),

187 X_WIN32(ERROR_ELEVATION_REQUIRED),

src/windows/inc/wslc_schema.h

+18

@@ -348,4 +348,22 @@ struct VolumeListEntry

348 NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(VolumeListEntry, Name, Driver, Mountpoint, Scope, Labels);

349 };

350

351 +struct EventActor

352 +{

353 + std::string ID;

354 + std::map<std::string, std::string> Attributes;

355 +

356 + NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(EventActor, ID, Attributes);

357 +};

358 +

359 +struct Event

360 +{

361 + std::string Type;

362 + std::string Action;

363 + EventActor Actor;

364 + std::int64_t time{};

365 +

366 + NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(Event, Type, Action, Actor, time);

367 +};

368 +

369 } // namespace wsl::windows::common::wslc_schema

src/windows/service/inc/wslc.idl

+26

@@ -671,6 +671,18 @@ typedef struct _WSLCListContainersOptions

671 ULONG FiltersCount;

672 } WSLCListContainersOptions;

673

674 +[

675 + uuid(7EC66D3B-D098-4D48-B69E-69166F6C4745),

676 + pointer_default(unique),

677 + object

678 +]

679 +interface IWSLCEventStream : IUnknown

680 +{

681 + // Blocks until the next matching event, the until-time is reached, or the session terminates,

682 + // then returns the event as a JSON object following the wslc_schema::Event format.

683 + HRESULT GetNext([out, string] LPSTR* EventJson);

684 +}

685 +

686 // Settings for IWSLCSession::Initialize - passed from service to per-user process

687 typedef struct _WSLCSessionInitSettings

688 {

@@ -708,6 +720,18 @@ interface IWSLCSession : IUnknown

720 // termination event has been signaled; before that the call fails.

721 HRESULT GetTerminationReason([out] WSLCVirtualMachineTerminationReason* Reason, [out] LPWSTR* Details);

722

723 + // Opens an event stream that mirrors `docker events`. The window and filters are captured by the

724 + // returned stream object; events are then pulled one at a time via IWSLCEventStream::GetNext.

725 + // SinceTime / UntilTime bound the window by event time, in seconds since the Unix epoch. SinceTime

726 + // is inclusive and UntilTime is exclusive. 0 means unbounded on that end.

727 + // Filters are key/value pairs. Values sharing a key are OR'd, distinct keys are AND'd.

728 + HRESULT GetEvents(

729 + [in] LONGLONG SinceTime,

730 + [in] LONGLONG UntilTime,

731 + [in, unique, size_is(FiltersCount)] const WSLCFilter* Filters,

732 + [in] ULONG FiltersCount,

733 + [out] IWSLCEventStream** Stream);

734 +

735 // Image management.

736 HRESULT PullImage([in] LPCSTR Image, [in, unique] LPCSTR RegistryAuthenticationInformation, [in, unique] IProgressCallback* ProgressCallback, [in, unique] IWarningCallback* WarningCallback);

737 HRESULT BuildImage([in] const WSLCBuildImageOptions* Options, [in, unique] IProgressCallback* ProgressCallback, [in, unique, system_handle(sh_event)] HANDLE CancelEvent);

@@ -885,3 +909,5 @@ cpp_quote("#define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILIT

909 // N.B. WSLC_E_VM_NOT_RUNNING is part of the plugin API contract and is also defined in WslPluginApi.h.

910 // The two definitions must stay in sync.

911 cpp_quote("#define WSLC_E_VM_NOT_RUNNING MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */")

912 +cpp_quote("#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */")

913 +cpp_quote("#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 18) /* 0x80040612 */")

src/windows/wslcsession/CMakeLists.txt

@@ -26,6 +26,7 @@ set(SOURCES

26 # Supporting classes

27 DockerEventTracker.cpp

28 DockerHTTPClient.cpp

29 + EventStore.cpp

30 IORelay.cpp

31 OptionParser.cpp

32 ServiceProcessLauncher.cpp

@@ -35,6 +36,7 @@ set(SOURCES

36 set(HEADERS

37 DockerEventTracker.h

38 DockerHTTPClient.h

39 + EventStore.h

40 IORelay.h

41 OptionParser.h

42 ServiceProcessLauncher.h

src/windows/wslcsession/DockerEventTracker.cpp

+43 -55

@@ -96,6 +96,7 @@ DockerEventTracker::~DockerEventTracker()

96 // N.B. No callback should be left when the tracker is destroyed.

97 WI_ASSERT(m_containerCallbacks.empty());

98 WI_ASSERT(m_volumeCallbacks.empty());

99 + WI_ASSERT(m_containerCreateCallbacks.empty());

100 }

101

102 void DockerEventTracker::OnEvent(const std::string_view& event)

@@ -125,33 +126,16 @@ void DockerEventTracker::OnEvent(const std::string_view& event)

126 if (typeStr == "container")

127 {

128 OnContainerEvent(parsed, actionStr, eventTime);

129 +

130 + if (actionStr == "create")

131 + {

132 + OnContainerCreated(parsed, eventTime);

133 + }

134 }

135 else if (typeStr == "volume")

136 {

137 OnVolumeEvent(parsed, actionStr, eventTime);

138 }

133 -

134 - // Track object creation for WaitForObjectCreated.

135 - auto actor = parsed.find("Actor");

136 - if (actor != parsed.end())

137 - {

138 - auto id = actor->find("ID");

139 - if (id != actor->end())

140 - {

141 - auto objectId = id->get<std::string>();

142 - if (actionStr == "create")

143 - {

144 - std::lock_guard lock{m_lock};

145 - m_createdObjects.insert(objectId);

146 - m_objectCreated.SetEvent();

147 - }

148 - else if (actionStr == "destroy")

149 - {

150 - std::lock_guard lock{m_lock};

151 - m_createdObjects.erase(objectId);

152 - }

153 - }

154 - }

139 }

140

141 void DockerEventTracker::OnContainerEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime)

@@ -159,6 +143,7 @@ void DockerEventTracker::OnContainerEvent(const nlohmann::json& parsed, const st

143 static std::map<std::string, ContainerEvent> events{

144 {"start", ContainerEvent::Start},

145 {"die", ContainerEvent::Stop},

146 + {"kill", ContainerEvent::Kill},

147 {"destroy", ContainerEvent::Destroy},

148 {"exec_die", ContainerEvent::ExecDied},

149 {"restart", ContainerEvent::Restart}};

@@ -241,30 +226,23 @@ void DockerEventTracker::OnVolumeEvent(const nlohmann::json& parsed, const std::

226 InvokeCallbacks(callbacks, [&](const VolumeCallback& e) { e.Callback(volumeName, it->second, eventTime); });

227 }

228

244 -void DockerEventTracker::WaitForObjectCreated(const std::string& ObjectId)

229 +void DockerEventTracker::OnContainerCreated(const nlohmann::json& parsed, std::int64_t eventTime)

230 {

246 - constexpr auto c_timeout = std::chrono::seconds{60};

231 + auto actor = parsed.find("Actor");

232 + THROW_HR_IF_MSG(E_INVALIDARG, actor == parsed.end(), "Missing Actor in container event");

233

248 - while (true)

249 - {

250 - {

251 - std::lock_guard lock{m_lock};

252 - if (m_createdObjects.contains(ObjectId))

253 - {

254 - return;

255 - }

234 + auto id = actor->find("ID");

235 + THROW_HR_IF_MSG(E_INVALIDARG, id == actor->end(), "Missing Actor.ID in container event");

236

257 - // Reset under the lock so a concurrent OnEvent() that runs after we release the lock

258 - // and before the wait can re-signal the event and unblock us.

259 - m_objectCreated.ResetEvent();

260 - }

237 + auto containerId = id->get<std::string>();

238

262 - THROW_HR_IF_MSG(

263 - HRESULT_FROM_WIN32(ERROR_TIMEOUT),

264 - !m_session.WaitForEventOrSessionTerminating(m_objectCreated.get(), c_timeout),

265 - "Timed out waiting for Docker create event for object '%hs'",

266 - ObjectId.c_str());

239 + std::vector<std::shared_ptr<ContainerCreateCallbackEntry>> callbacks;

240 + {

241 + std::lock_guard lock{m_lock};

242 + callbacks = m_containerCreateCallbacks;

243 }

244 +

245 + InvokeCallbacks(callbacks, [&](const ContainerCreateCallbackEntry& e) { e.Callback(containerId, eventTime); });

246 }

247

248 DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterContainerStateUpdates(

@@ -302,6 +280,17 @@ DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterVolumeUpd

280 return EventTrackingReference{this, id};

281 }

282

283 +DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterContainerCreate(ContainerCreateCallback&& Callback) noexcept

284 +{

285 + auto id = m_callbackId++;

286 + auto entry = std::make_shared<ContainerCreateCallbackEntry>(id, std::move(Callback));

287 +

288 + std::lock_guard lock{m_lock};

289 + m_containerCreateCallbacks.emplace_back(std::move(entry));

290 +

291 + return EventTrackingReference{this, id};

292 +}

293 +

294 void DockerEventTracker::UnregisterCallback(size_t Id) noexcept

295 {

296 std::shared_ptr<CallbackRegistration> registration;

@@ -311,22 +300,21 @@ void DockerEventTracker::UnregisterCallback(size_t Id) noexcept

300

301 auto matches = [Id](const auto& e) { return e->CallbackId == Id; };

302

314 - // Try container callbacks first, then volume callbacks.

315 - if (auto container = std::ranges::find_if(m_containerCallbacks, matches); container != m_containerCallbacks.end())

316 - {

317 - registration = std::move(*container);

318 - m_containerCallbacks.erase(container);

319 - }

320 - else

321 - {

322 - auto volume = std::ranges::find_if(m_volumeCallbacks, matches);

323 - WI_ASSERT(volume != m_volumeCallbacks.end());

324 -

325 - if (volume != m_volumeCallbacks.end())

303 + auto take = [&](auto& Callbacks) {

304 + auto entry = std::ranges::find_if(Callbacks, matches);

305 + if (entry == Callbacks.end())

306 {

327 - registration = std::move(*volume);

328 - m_volumeCallbacks.erase(volume);

307 + return false;

308 }

309 +

310 + registration = std::move(*entry);

311 + Callbacks.erase(entry);

312 + return true;

313 + };

314 +

315 + if (!take(m_containerCallbacks) && !take(m_volumeCallbacks) && !take(m_containerCreateCallbacks))

316 + {

317 + WI_ASSERT(false);

318 }

319 }

320

src/windows/wslcsession/DockerEventTracker.h

+19 -6

@@ -30,7 +30,8 @@ enum class ContainerEvent

30 Stop,

31 Exit,

32 Destroy,

33 - ExecDied

33 + ExecDied,

34 + Kill

35 };

37 enum class VolumeEvent

@@ -64,6 +65,7 @@ public:

66 using ContainerStateChangeCallback = std::function<void(ContainerEvent, std::optional<int>, std::int64_t)>;

67 using VolumeEventCallback = std::function<void(const std::string&, VolumeEvent, std::int64_t)>;

68 + using ContainerCreateCallback = std::function<void(const std::string& ContainerId, std::int64_t Time)>;

70 explicit DockerEventTracker(WSLCSession& session);

71 ~DockerEventTracker();

@@ -76,13 +78,16 @@ public:

78 EventTrackingReference RegisterContainerStateUpdates(const std::string& ContainerId, ContainerStateChangeCallback&& Callback) noexcept;

79 EventTrackingReference RegisterExecStateUpdates(const std::string& ContainerId, const std::string& ExecId, ContainerStateChangeCallback&& Callback) noexcept;

80 EventTrackingReference RegisterVolumeUpdates(VolumeEventCallback&& Callback) noexcept;

79 - void UnregisterCallback(size_t Id) noexcept;

81 - void WaitForObjectCreated(const std::string& ObjectId);

82 + // Invoked for every container create event, after the per-container state callbacks. Unlike those,

83 + // this isn't keyed by container id, because the id isn't known until Docker assigns it.

84 + EventTrackingReference RegisterContainerCreate(ContainerCreateCallback&& Callback) noexcept;

85 + void UnregisterCallback(size_t Id) noexcept;

87 private:

88 void OnEvent(const std::string_view& event);

89 void OnContainerEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime);

90 + void OnContainerCreated(const nlohmann::json& parsed, std::int64_t eventTime);

91 void OnVolumeEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime);

93 // Callbacks are invoked without holding m_lock so that a callback can register or unregister callbacks, and so

@@ -125,8 +130,19 @@ private:

130 const VolumeEventCallback Callback;

131 };

132

133 + struct ContainerCreateCallbackEntry : CallbackRegistration

134 + {

135 + ContainerCreateCallbackEntry(size_t Id, ContainerCreateCallback&& Callback) :

136 + CallbackRegistration(Id), Callback(std::move(Callback))

137 + {

138 + }

139 +

140 + const ContainerCreateCallback Callback;

141 + };

142 +

143 _Guarded_by_(m_lock) std::vector<std::shared_ptr<ContainerCallback>> m_containerCallbacks;

144 _Guarded_by_(m_lock) std::vector<std::shared_ptr<VolumeCallback>> m_volumeCallbacks;

145 + _Guarded_by_(m_lock) std::vector<std::shared_ptr<ContainerCreateCallbackEntry>> m_containerCreateCallbacks;

146

147 // Invokes a snapshot of callbacks taken under m_lock, skipping registrations that have since been unregistered.

148 template <typename TCallback, typename TInvoke>

@@ -142,9 +158,6 @@ private:

158 }

159 }

160

145 - _Guarded_by_(m_lock) std::unordered_set<std::string> m_createdObjects;

146 - _Guarded_by_(m_lock) wil::unique_event m_objectCreated { wil::EventOptions::ManualReset };

147 -

161 WSLCSession& m_session;

162 std::mutex m_lock;

163 std::atomic<size_t> m_callbackId{0};

src/windows/wslcsession/EventStore.cpp new

+259

@@ -0,0 +1,259 @@

2 +

3 +#include "precomp.h"

4 +#include "EventStore.h"

5 +#include "WSLCSession.h"

6 +#include "WSLCExecutionContext.h"

7 +#include <chrono>

8 +

9 +using wsl::shared::Localization;

10 +

11 +namespace wsl::windows::service::wslc {

12 +

13 +namespace {

14 +

15 + std::optional<std::chrono::sys_seconds> ToTimeBound(int64_t TimeSeconds)

16 + {

17 + if (TimeSeconds == 0)

18 + {

19 + return std::nullopt;

20 + }

21 +

22 + // Waiting on a bound converts it to the system clock's 100ns ticks, which a far-future second would overflow.

23 + constexpr auto c_maxBound = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::time_point::max());

24 + return std::min(std::chrono::sys_seconds{std::chrono::seconds{TimeSeconds}}, c_maxBound);

25 + }

26 +

27 +} // namespace

28 +

29 +void EventStore::Append(wsl::windows::common::wslc_schema::Event Event)

30 +{

31 + std::lock_guard lock(m_lock);

32 +

33 + // Events are recorded in Docker's delivery order, which is also timestamp order. Subscribers rely on

34 + // this: they resume from a sequence number, so an out-of-order event could never be inserted where it

35 + // belongs without hiding it from readers that already moved past that point.

36 + WI_ASSERT(m_events.empty() || m_events.back().time <= Event.time);

37 +

38 + m_events.push_back(std::move(Event));

39 +

40 + if (m_events.size() > c_eventRingCapacity)

41 + {

42 + m_events.pop_front();

43 + ++m_firstSequenceNumber;

44 + }

45 +

46 + m_updated.notify_all();

47 +}

48 +

49 +void EventStore::Record(std::string&& Type, std::string&& Action, const std::string& ActorId, std::map<std::string, std::string> ActorAttributes, std::int64_t Time) noexcept

50 +try

51 +{

52 + wsl::windows::common::wslc_schema::Event event;

53 + event.Type = std::move(Type);

54 + event.Action = std::move(Action);

55 + event.Actor.ID = ActorId;

56 + event.Actor.Attributes = std::move(ActorAttributes);

57 + event.time = Time;

58 +

59 + Append(std::move(event));

60 +}

61 +CATCH_LOG()

62 +

63 +namespace {

64 +

65 + // Values sharing a key are OR'd, distinct keys are AND'd. Unrecognized keys are ignored.

66 + bool EventMatchesFilters(const wsl::windows::common::wslc_schema::Event& event, const std::map<std::string, std::vector<std::string>>& filters)

67 + {

68 + for (const auto& [key, values] : filters)

69 + {

70 + if (key == "type")

71 + {

72 + if (!std::ranges::any_of(values, [&](const std::string& v) { return event.Type == v; }))

73 + {

74 + return false;

75 + }

76 + }

77 + else if (key == "event")

78 + {

79 + if (!std::ranges::any_of(values, [&](const std::string& v) { return event.Action == v; }))

80 + {

81 + return false;

82 + }

83 + }

84 + else if (key == "container")

85 + {

86 + if (event.Type != "container" ||

87 + !std::ranges::any_of(values, [&](const std::string& v) { return event.Actor.ID == v; }))

88 + {

89 + return false;

90 + }

91 + }

92 + else if (key == "image")

93 + {

94 + if (event.Type != "image" || !std::ranges::any_of(values, [&](const std::string& v) { return event.Actor.ID == v; }))

95 + {

96 + return false;

97 + }

98 + }

99 + }

100 + return true;

101 + }

102 +

103 +} // namespace

104 +

105 +Microsoft::WRL::ComPtr<IWSLCEventStream> EventStore::CreateStream(

106 + Microsoft::WRL::ComPtr<WSLCSession> Session, int64_t SinceTime, int64_t UntilTime, std::map<std::string, std::vector<std::string>> Filters)

107 +{

108 + // Zero means unbounded on that end, so it never makes the window run backwards.

109 + THROW_HR_WITH_USER_ERROR_IF(

110 + E_INVALIDARG,

111 + Localization::MessageWslcEventsInvalidTimeWindow(SinceTime, UntilTime),

112 + SinceTime < 0 || UntilTime < 0 || (SinceTime != 0 && UntilTime != 0 && SinceTime > UntilTime));

113 +

114 + Microsoft::WRL::ComPtr<EventStream> stream;

115 + THROW_IF_FAILED(Microsoft::WRL::MakeAndInitialize<EventStream>(&stream, std::move(Session), this, SinceTime, UntilTime, std::move(Filters)));

116 +

117 + return stream;

118 +}

119 +

120 +std::optional<wsl::windows::common::wslc_schema::Event> EventStore::GetLockHeld(uint64_t SequenceNumber)

121 +{

122 + // Callers resync a lagging reader before reaching here, so the requested event is never evicted.

123 + WI_ASSERT(SequenceNumber >= m_firstSequenceNumber);

124 +

125 + const uint64_t index = SequenceNumber - m_firstSequenceNumber;

126 + if (index >= m_events.size())

127 + {

128 + return std::nullopt;

129 + }

130 +

131 + return m_events[index];

132 +}

133 +

134 +bool EventStore::WaitForEvent(std::unique_lock<std::mutex>& Lock, uint64_t SequenceNumber, std::optional<std::chrono::sys_seconds> Until)

135 +{

136 + // Ready once the reader's event is buffered, its slot is evicted, or the session terminates.

137 + // Eviction while parked wakes us too, so the caller reports the gap on its next pass.

138 + const auto ready = [&] { return m_terminating || SequenceNumber < m_firstSequenceNumber + m_events.size(); };

139 +

140 + if (Until.has_value())

141 + {

142 + if (!m_updated.wait_until(Lock, Until.value(), ready))

143 + {

144 + return false;

145 + }

146 + }

147 + else

148 + {

149 + m_updated.wait(Lock, ready);

150 + }

151 +

152 + THROW_HR_IF(E_ABORT, m_terminating);

153 + return true;

154 +}

155 +

156 +std::optional<wsl::windows::common::wslc_schema::Event> EventStore::Get(

157 + std::optional<uint64_t>& SequenceNumber,

158 + std::optional<std::chrono::sys_seconds> Since,

159 + std::optional<std::chrono::sys_seconds> Until,

160 + const std::map<std::string, std::vector<std::string>>& Filters)

161 +{

162 + std::unique_lock lock(m_lock);

163 +

164 + // Position the reader. A first read (no sequence number yet) starts at the oldest buffered

165 + // event

166 + SequenceNumber = SequenceNumber.value_or(m_firstSequenceNumber);

167 +

168 + while (true)

169 + {

170 + // A reader that has fallen behind the ring missed events to eviction: reset it so the

171 + // next call starts fresh at the oldest buffered event, and report the gap.

172 + if (SequenceNumber.value() < m_firstSequenceNumber)

173 + {

174 + SequenceNumber = std::nullopt;

175 + THROW_HR(WSLC_E_EVENTS_LOST);

176 + }

177 +

178 + if (!WaitForEvent(lock, SequenceNumber.value(), Until))

179 + {

180 + // The until window elapsed with no further event: the stream is finished.

181 + return std::nullopt;

182 + }

183 +

184 + // Evicted while parked: loop back to reset and report the gap.

185 + // TODO: A burst of more than c_eventRingCapacity events between the wake and reacquiring the

186 + // lock can evict this reader's event before it is read, forcing a WSLC_E_EVENTS_LOST. Redesign

187 + // so that every parked reader is guaranteed to observe an event before the next write can evict

188 + // it.

189 + if (SequenceNumber.value() < m_firstSequenceNumber)

190 + {

191 + continue;

192 + }

193 +

194 + const auto event = GetLockHeld(SequenceNumber.value()).value();

195 + const std::chrono::sys_seconds eventTime{std::chrono::seconds{event.time}};

196 +

197 + // Advance in delivery order before applying the time window.

198 + SequenceNumber.value()++;

199 +

200 + // Events are appended in non-decreasing timestamp order (see Append()), so once we reach the

201 + // exclusive Until bound, the stream is finished.

202 + if (Until.has_value() && eventTime >= Until.value())

203 + {

204 + return std::nullopt;

205 + }

206 +

207 + // Return the event if it falls within the since-bound and matches the caller's filters;

208 + // otherwise loop to skip it.

209 + if ((!Since.has_value() || eventTime >= Since.value()) && EventMatchesFilters(event, Filters))

210 + {

211 + return event;

212 + }

213 + }

214 +}

215 +

216 +void EventStore::OnSessionTerminating()

217 +{

218 + {

219 + std::lock_guard lock(m_lock);

220 + m_terminating = true;

221 + }

222 +

223 + m_updated.notify_all();

224 +}

225 +

226 +HRESULT EventStream::RuntimeClassInitialize(

227 + Microsoft::WRL::ComPtr<WSLCSession> Session,

228 + EventStore* Store,

229 + int64_t SinceTime,

230 + int64_t UntilTime,

231 + std::map<std::string, std::vector<std::string>> Filters)

232 +{

233 + m_session = std::move(Session);

234 + m_store = Store;

235 + m_since = ToTimeBound(SinceTime);

236 + m_until = ToTimeBound(UntilTime);

237 + m_filters = std::move(Filters);

238 + return S_OK;

239 +}

240 +

241 +HRESULT EventStream::GetNext(LPSTR* EventJson)

242 +try

243 +{

244 + RETURN_HR_IF_NULL(E_POINTER, EventJson);

245 + *EventJson = nullptr;

246 +

247 + std::lock_guard lock(m_lock);

248 + const auto event = m_store->Get(m_nextSequenceNumber, m_since, m_until, m_filters);

249 + if (!event.has_value())

250 + {

251 + return WSLC_E_EVENT_STREAM_FINISHED;

252 + }

253 +

254 + *EventJson = wil::make_unique_ansistring<wil::unique_cotaskmem_ansistring>(wsl::shared::ToJson(event.value()).c_str()).release();

255 + return S_OK;

256 +}

257 +CATCH_RETURN();

258 +

259 +} // namespace wsl::windows::service::wslc

src/windows/wslcsession/EventStore.h new

+88

@@ -0,0 +1,88 @@

2 +

3 +#pragma once

4 +

5 +#include <chrono>

6 +#include <condition_variable>

7 +#include <cstdint>

8 +#include <deque>

9 +#include <map>

10 +#include <mutex>

11 +#include <optional>

12 +#include <string>

13 +#include <vector>

14 +#include "wslc.h"

15 +#include "wslc_schema.h"

16 +

17 +namespace wsl::windows::service::wslc {

18 +

19 +class WSLCSession;

20 +

21 +class EventStore

22 +{

23 +public:

24 + static constexpr size_t c_eventRingCapacity = 256;

25 +

26 + void Record(std::string&& Type, std::string&& Action, const std::string& ActorId, std::map<std::string, std::string> ActorAttributes, std::int64_t Time) noexcept;

27 +

28 + Microsoft::WRL::ComPtr<IWSLCEventStream> CreateStream(

29 + Microsoft::WRL::ComPtr<WSLCSession> Session, int64_t SinceTime, int64_t UntilTime, std::map<std::string, std::vector<std::string>> Filters);

30 +

31 + // Returns the next event at or after SequenceNumber that falls within [Since, Until) and matches

32 + // Filters, advancing SequenceNumber past it. A nullopt SequenceNumber starts a fresh reader at the

33 + // oldest buffered event (no gap is reported). If the reader has since fallen behind the ring,

34 + // resyncs SequenceNumber to the oldest buffered event and throws WSLC_E_EVENTS_LOST. Returns

35 + // nullopt once the Until window has closed.

36 + std::optional<wsl::windows::common::wslc_schema::Event> Get(

37 + std::optional<uint64_t>& SequenceNumber,

38 + std::optional<std::chrono::sys_seconds> Since,

39 + std::optional<std::chrono::sys_seconds> Until,

40 + const std::map<std::string, std::vector<std::string>>& Filters);

41 +

42 + void OnSessionTerminating();

43 +

44 +private:

45 + void Append(wsl::windows::common::wslc_schema::Event Event);

46 +

47 + // Blocks until the event at SequenceNumber is buffered, its slot is evicted, or the session

48 + // terminates. Returns false only when Until elapsed with no event ready. Throws E_ABORT if the

49 + // session terminated while waiting.

50 + bool WaitForEvent(std::unique_lock<std::mutex>& Lock, uint64_t SequenceNumber, std::optional<std::chrono::sys_seconds> Until);

51 +

52 + std::optional<wsl::windows::common::wslc_schema::Event> GetLockHeld(uint64_t SequenceNumber);

53 +

54 + std::mutex m_lock;

55 + std::condition_variable m_updated;

56 +

57 + _Guarded_by_(m_lock) std::deque<wsl::windows::common::wslc_schema::Event> m_events;

58 + _Guarded_by_(m_lock) uint64_t m_firstSequenceNumber = 1;

59 +

60 + _Guarded_by_(m_lock) bool m_terminating = false;

61 +};

62 +

63 +class EventStream

64 + : public Microsoft::WRL::RuntimeClass<Microsoft::WRL::RuntimeClassFlags<Microsoft::WRL::ClassicCom>, IWSLCEventStream, IFastRundown>

65 +{

66 +public:

67 + HRESULT RuntimeClassInitialize(

68 + Microsoft::WRL::ComPtr<WSLCSession> Session,

69 + EventStore* Store,

70 + int64_t SinceTime,

71 + int64_t UntilTime,

72 + std::map<std::string, std::vector<std::string>> Filters);

73 +

74 + IFACEMETHOD(GetNext)(_Outptr_result_z_ LPSTR* EventJson) override;

75 +

76 +private:

77 + Microsoft::WRL::ComPtr<WSLCSession> m_session;

78 + EventStore* m_store = nullptr;

79 +

80 + std::optional<std::chrono::sys_seconds> m_since;

81 + std::optional<std::chrono::sys_seconds> m_until;

82 + std::map<std::string, std::vector<std::string>> m_filters;

83 +

84 + std::mutex m_lock;

85 + _Guarded_by_(m_lock) std::optional<uint64_t> m_nextSequenceNumber;

86 +};

87 +

88 +} // namespace wsl::windows::service::wslc

src/windows/wslcsession/WSLCContainer.cpp

+53 -13

@@ -474,6 +474,21 @@ WSLCContainerState DockerStateToWSLCState(ContainerState state)

474 }

475 }

476

477 +std::string WSLCStateToEventAction(WSLCContainerState state)

478 +{

479 + switch (state)

480 + {

481 + case WslcContainerStateRunning:

482 + return "start";

483 + case WslcContainerStateExited:

484 + return "stop";

485 + case WslcContainerStateDeleted:

486 + return "destroy";

487 + default:

488 + WI_ASSERT(false);

489 + return "unknown";

490 + }

491 +}

492 std::string CleanContainerName(const std::string& name)

493 {

494 // Docker container names have a leading '/', strip it.

@@ -814,6 +829,7 @@ WSLCContainerImpl::WSLCContainerImpl(

829 std::vector<ContainerPortMapping>&& ports,

830 std::map<std::string, std::string>&& labels,

831 std::function<void(const WSLCContainerImpl*)>&& onDeleted,

832 + EventStore& eventStore,

833 WSLCContainerState InitialState,

834 std::int64_t CreatedAt,

835 WSLCProcessFlags InitProcessFlags,

@@ -832,6 +848,7 @@ WSLCContainerImpl::WSLCContainerImpl(

848 m_comWrapper(wil::MakeOrThrow<WSLCContainer>(wslcSession, std::move(onDeleted))),

849 m_containerEvents(runtime.Events().RegisterContainerStateUpdates(

850 m_id, std::bind(&WSLCContainerImpl::OnEvent, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),

851 + m_eventStore(eventStore),

852 m_state(InitialState),

853 m_createdAt(CreatedAt),

854 m_initProcessFlags(InitProcessFlags),

@@ -1224,12 +1241,34 @@ __requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::CompleteTransitio

1241 transition->Completed.SetEvent();

1242 }

1243

1244 +void WSLCContainerImpl::RecordEvent(std::string&& Action, std::int64_t Time, std::optional<int> ExitCode) noexcept

1245 +try

1246 +{

1247 + auto attributes = StripInternalLabels(m_labels);

1248 + attributes["name"] = m_name;

1249 + attributes["image"] = m_image;

1250 +

1251 + if (ExitCode.has_value())

1252 + {

1253 + attributes["exitCode"] = std::to_string(ExitCode.value());

1254 + }

1255 +

1256 + m_eventStore.Record("container", std::move(Action), m_id, std::move(attributes), Time);

1257 +}

1258 +CATCH_LOG()

1259 +

1260 void WSLCContainerImpl::OnEvent(ContainerEvent event, std::optional<int> exitCode, std::int64_t eventTime) noexcept

1261 {

1262 // Either owner may disconnect the COM wrapper, so both must outlive m_lock.

1263 unique_com_disconnect comWrapper;

1264 std::shared_ptr<StateTransition> transition;

1265

1266 + if (event == ContainerEvent::Kill)

1267 + {

1268 + RecordEvent("kill", eventTime);

1269 + return;

1270 + }

1271 +

1272 {

1273 auto lifecycleLock = m_lifecycleLock.lock_exclusive();

1274 auto lock = m_lock.lock_exclusive();

@@ -1396,7 +1435,7 @@ void WSLCContainerImpl::Stop(WSLCSignal Signal, LONG TimeoutSeconds, bool Kill)

1435 }

1436 }

1437

1399 -__requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::OnStopped(int exitCode, std::optional<std::int64_t> stopTimestamp)

1438 +__requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::OnStopped(int exitCode, std::int64_t stopTime)

1439 {

1440 auto transition = m_transition;

1441

@@ -1427,7 +1466,7 @@ __requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::OnStopped(int exi

1466 // Ignore duplicate or late Stop events so they do not overwrite an already committed state.

1467 if (m_state == WslcContainerStateRunning)

1468 {

1430 - CommitState(WslcContainerStateExited, stopTimestamp);

1469 + CommitState(WslcContainerStateExited, stopTime, exitCode);

1470 }

1471

1472 std::exception_ptr transitionException;

@@ -2048,11 +2087,11 @@ std::shared_ptr<WSLCContainerImpl> WSLCContainerImpl::Create(

2087 WSLCSessionRuntime& runtime,

2088 IWSLCPluginNotifier* pluginNotifier,

2089 const std::unordered_map<std::string, NetworkEntry>& sessionNetworks,

2051 - std::function<void(const WSLCContainerImpl*)>&& OnDeleted)

2090 + std::function<void(const WSLCContainerImpl*)>&& OnDeleted,

2091 + EventStore& eventStore)

2092 {

2093 auto& virtualMachine = runtime.Vm();

2094 auto& DockerClient = runtime.Docker();

2055 - auto& EventTracker = runtime.Events();

2095 const auto mounts = ConvertAndValidateMounts(containerOptions);

2096

2097 common::docker_schema::CreateContainer request;

@@ -2486,11 +2525,6 @@ std::shared_ptr<WSLCContainerImpl> WSLCContainerImpl::Create(

2525 name.c_str());

2526 }

2527

2489 - // Wait for the container create event to be delivered on the Docker event stream so that

2490 - // any events for objects created for the container (e.g. volumes) are delivered before we return

2491 - // from this function.

2492 - EventTracker.WaitForObjectCreated(result.Id);

2493 -

2528 // Collect the names of referenced docker named volumes so Start() can verify

2529 // they are available before running the container.

2530 std::vector<std::string> namedVolumes;

@@ -2509,6 +2543,7 @@ std::shared_ptr<WSLCContainerImpl> WSLCContainerImpl::Create(

2543 }

2544

2545 auto mergedLabels = StripInternalLabels(std::move(inspectData.Config.Labels));

2546 + const auto createdAt = wsl::windows::common::timestamp::Rfc3339ToEpoch(inspectData.Created);

2547

2548 auto container = std::make_shared<WSLCContainerImpl>(

2549 wslcSession,

@@ -2523,8 +2558,9 @@ std::shared_ptr<WSLCContainerImpl> WSLCContainerImpl::Create(

2558 std::move(mappedPorts),

2559 std::move(mergedLabels),

2560 std::move(OnDeleted),

2561 + eventStore,

2562 WslcContainerStateCreated,

2527 - wsl::windows::common::timestamp::Rfc3339ToEpoch(inspectData.Created),

2563 + createdAt,

2564 containerOptions.InitProcessOptions.Flags,

2565 containerOptions.Flags);

2566

@@ -2539,7 +2575,8 @@ std::shared_ptr<WSLCContainerImpl> WSLCContainerImpl::Open(

2575 WSLCSession& wslcSession,

2576 WSLCSessionRuntime& runtime,

2577 IWSLCPluginNotifier* pluginNotifier,

2542 - std::function<void(const WSLCContainerImpl*)>&& OnDeleted)

2578 + std::function<void(const WSLCContainerImpl*)>&& OnDeleted,

2579 + EventStore& eventStore)

2580 {

2581 auto& virtualMachine = runtime.Vm();

2582 auto& DockerClient = runtime.Docker();

@@ -2611,6 +2648,7 @@ std::shared_ptr<WSLCContainerImpl> WSLCContainerImpl::Open(

2648 std::move(ports),

2649 std::move(labels),

2650 std::move(OnDeleted),

2651 + eventStore,

2652 DockerStateToWSLCState(dockerContainer.State),

2653 dockerContainer.Created,

2654 metadata.InitProcessFlags,

@@ -2917,7 +2955,7 @@ __requires_exclusive_lock_held(m_lock) unique_com_disconnect WSLCContainerImpl::

2955 return unique_com_disconnect{std::exchange(m_comWrapper, nullptr)};

2956 }

2957

2920 -__requires_lock_held(m_lock) void WSLCContainerImpl::CommitState(WSLCContainerState State, std::optional<std::int64_t> stateChangedAt) noexcept

2958 +__requires_lock_held(m_lock) void WSLCContainerImpl::CommitState(WSLCContainerState State, std::int64_t Time, std::optional<int> ExitCode) noexcept

2959 {

2960 // N.B. A deleted container cannot transition back to any other state.

2961 WI_ASSERT(m_state != WslcContainerStateDeleted);

@@ -2930,7 +2968,9 @@ __requires_lock_held(m_lock) void WSLCContainerImpl::CommitState(WSLCContainerSt

2968

2969 m_state = State;

2970 m_stateGeneration++;

2933 - m_stateChangedAt = stateChangedAt.value_or(static_cast<std::int64_t>(std::time(nullptr)));

2971 + m_stateChangedAt = Time;

2972 +

2973 + RecordEvent(WSLCStateToEventAction(State), Time, ExitCode);

2974

2975 // Keep the VM alive while this container is Running and release the hold once it leaves that

2976 // state, even when no client holds the wrapper (e.g. a detached `run -d` container). Dropping

src/windows/wslcsession/WSLCContainer.h

+13 -4

@@ -35,6 +35,7 @@ class WSLCContainer;

35 class WSLCSession;

36 class WSLCSessionRuntime;

37 class WSLCVolumes;

38 +class EventStore;

40 class unique_com_disconnect

41 {

@@ -85,6 +86,7 @@ public:

86 std::vector<ContainerPortMapping>&& ports,

87 std::map<std::string, std::string>&& labels,

88 std::function<void(const WSLCContainerImpl*)>&& OnDeleted,

89 + EventStore& eventStore,

90 WSLCContainerState InitialState,

91 std::int64_t CreatedAt,

92 WSLCProcessFlags InitProcessFlags,

@@ -123,7 +125,7 @@ public:

125 // Re-registers a stopped container's VM-scoped port allocations against the restarted VM.

126 void RecoverPorts(const common::docker_schema::ContainerInfo& dockerContainer);

127

126 - __requires_lock_held(m_lock) void CommitState(WSLCContainerState State, std::optional<std::int64_t> stateChangedAt = std::nullopt) noexcept;

128 + __requires_lock_held(m_lock) void CommitState(WSLCContainerState State, std::int64_t Time, std::optional<int> ExitCode = std::nullopt) noexcept;

129

130 const std::string& ID() const noexcept;

131

@@ -141,14 +143,20 @@ public:

143 WSLCSessionRuntime& runtime,

144 IWSLCPluginNotifier* pluginNotifier,

145 const std::unordered_map<std::string, NetworkEntry>& SessionNetworks,

144 - std::function<void(const WSLCContainerImpl*)>&& OnDeleted);

146 + std::function<void(const WSLCContainerImpl*)>&& OnDeleted,

147 + EventStore& eventStore);

148

149 static std::shared_ptr<WSLCContainerImpl> Open(

150 const common::docker_schema::ContainerInfo& DockerContainer,

151 WSLCSession& wslcSession,

152 WSLCSessionRuntime& runtime,

153 IWSLCPluginNotifier* pluginNotifier,

151 - std::function<void(const WSLCContainerImpl*)>&& OnDeleted);

154 + std::function<void(const WSLCContainerImpl*)>&& OnDeleted,

155 + EventStore& eventStore);

156 +

157 + // Appends an event for this container to the session's event stream. Must be called from the Docker

158 + // event stream thread so that recorded events keep Docker's delivery order.

159 + void RecordEvent(std::string&& Action, std::int64_t Time, std::optional<int> ExitCode = std::nullopt) noexcept;

160

161 private:

162 enum class TransitionKind

@@ -197,7 +205,7 @@ private:

205 __requires_exclusive_lock_held(m_lock) void ReleaseProcesses();

206 __requires_exclusive_lock_held(m_lock) [[nodiscard]] unique_com_disconnect PrepareDisconnectComWrapper();

207

200 - __requires_exclusive_lock_held(m_lock) void OnStopped(int exitCode, std::optional<std::int64_t> stopTimestamp);

208 + __requires_exclusive_lock_held(m_lock) void OnStopped(int exitCode, std::int64_t stopTime);

209

210 void SetExitCode(int ExitCode) noexcept;

211 void SignalInitProcessExit() noexcept;

@@ -252,6 +260,7 @@ private:

260 std::map<std::string, std::string> m_labels;

261 Microsoft::WRL::ComPtr<WSLCContainer> m_comWrapper;

262 DockerEventTracker::EventTrackingReference m_containerEvents;

263 + EventStore& m_eventStore;

264 std::string m_networkMode;

265

266 // Held (non-empty) exactly while the container is Running so the session's VM stays alive even

src/windows/wslcsession/WSLCProcessControl.cpp

+1 -1

@@ -172,7 +172,7 @@ void DockerExecProcessControl::SetExitCode(int ExitCode)

172 }

173 }

174

175 -void DockerExecProcessControl::OnEvent(ContainerEvent Event, std::optional<int> ExitCode, std::int64_t /*eventTime*/)

175 +void DockerExecProcessControl::OnEvent(ContainerEvent Event, std::optional<int> ExitCode, std::int64_t)

176 {

177 if (Event == ContainerEvent::ExecDied && !m_exitEvent.is_signaled())

178 {

src/windows/wslcsession/WSLCSession.cpp

+135 -7

@@ -41,6 +41,7 @@ using wsl::windows::service::wslc::WSLCVirtualMachine;

41 constexpr auto c_containerdSocket = "/run/containerd/containerd.sock";

42 constexpr auto c_storageVhdFilename = wsl::windows::wslc::DefaultStorageVhdName;

43 constexpr uint32_t c_progressPrecision = 4;

44 +constexpr auto c_containerCreateEventTimeout = std::chrono::seconds{60};

46 // Default grace period to keep an otherwise-idle VM running before tearing it down (used when the

47 // session's IdleTimeoutSec setting is 0/unset). This avoids thrashing the VM (repeated

@@ -499,6 +500,9 @@ try

500

501 m_runtime.Initialize(m_vmFactoryGitCookie, m_git, &m_settings, idleGracePeriod, std::move(sessionContext), std::move(hooks));

502

503 + m_containerEventTracking = m_runtime.Events().RegisterContainerCreate(

504 + std::bind(&WSLCSession::OnContainerCreated, this, std::placeholders::_1, std::placeholders::_2));

505 +

506 return S_OK;

507 }

508 CATCH_RETURN()

@@ -2312,7 +2316,9 @@ void WSLCSession::CreateContainerImpl(const WSLCContainerOptions* containerOptio

2316

2317 try

2318 {

2315 - std::scoped_lock lock(m_containersLock, m_networksLock);

2319 + std::unique_lock containersLock{m_containersLock};

2320 + WaitForConflictingCreateToComplete(containersLock);

2321 + std::unique_lock networksLock{m_networksLock};

2322

2323 // Generate a unique container name if the user didn't provide one.

2324 std::string containerName;

@@ -2350,13 +2356,24 @@ void WSLCSession::CreateContainerImpl(const WSLCContainerOptions* containerOptio

2356 m_runtime,

2357 m_pluginNotifier.get(),

2358 m_networks,

2353 - std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1));

2359 + std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1),

2360 + m_eventStore);

2361

2355 - // Key the map by Docker's container ID, which is set in the WSLCContainerImpl constructor and stable for its lifetime.

2356 - auto [it, inserted] = m_containers.emplace(container->ID(), std::move(container));

2357 - WI_ASSERT(inserted);

2362 + auto pendingCreate = StartPendingCreate(container);

2363 +

2364 + containersLock.unlock();

2365 + networksLock.unlock();

2366

2359 - it->second->CopyTo(Container);

2367 + // m_pendingCreate is published under m_containersLock before the event thread can observe it, so

2368 + // OnContainerCreated() is guaranteed to complete this create unless the session tears down first.

2369 + WaitForPendingCreateCompletion(pendingCreate);

2370 +

2371 + if (pendingCreate->Exception)

2372 + {

2373 + std::rethrow_exception(pendingCreate->Exception);

2374 + }

2375 +

2376 + container->CopyTo(Container);

2377 }

2378 catch (const DockerHTTPException& e)

2379 {

@@ -2372,6 +2389,97 @@ void WSLCSession::CreateContainerImpl(const WSLCContainerOptions* containerOptio

2389 }

2390 }

2391

2392 +__requires_lock_held(m_containersLock) std::shared_ptr<WSLCSession::PendingContainerCreate> WSLCSession::StartPendingCreate(std::shared_ptr<WSLCContainerImpl> Container)

2393 +{

2394 + WI_ASSERT(!m_pendingCreate);

2395 +

2396 + m_pendingCreate = std::make_shared<PendingContainerCreate>();

2397 + m_pendingCreate->Container = std::move(Container);

2398 +

2399 + return m_pendingCreate;

2400 +}

2401 +

2402 +void WSLCSession::WaitForPendingCreateCompletion(const std::shared_ptr<PendingContainerCreate>& PendingCreate)

2403 +{

2404 + auto io = CreateIOContext();

2405 + io.AddHandle(std::make_unique<io::EventHandle>(PendingCreate->Completed.get()));

2406 +

2407 + try

2408 + {

2409 + io.Run(c_containerCreateEventTimeout);

2410 + }

2411 + catch (...)

2412 + {

2413 + if (wil::ResultFromCaughtException() != HRESULT_FROM_WIN32(ERROR_TIMEOUT))

2414 + {

2415 + throw;

2416 + }

2417 +

2418 + // Fail this create rather than leaving m_pendingCreate set, which would wedge every later one.

2419 + // Any container docker did manage to create is left behind; the same broken event stream makes

2420 + // deleting it unreliable, and a late create event is ignored once m_pendingCreate is cleared.

2421 + std::lock_guard containersLock{m_containersLock};

2422 + if (m_pendingCreate == PendingCreate)

2423 + {

2424 + CompletePendingCreate(PendingCreate, std::current_exception());

2425 + }

2426 + }

2427 +

2428 + WI_ASSERT(PendingCreate->Completed.is_signaled());

2429 +}

2430 +

2431 +__requires_lock_held(m_containersLock) void WSLCSession::CompletePendingCreate(

2432 + const std::shared_ptr<PendingContainerCreate>& PendingCreate, std::exception_ptr Exception) noexcept

2433 +{

2434 + WI_ASSERT(m_pendingCreate == PendingCreate);

2435 + PendingCreate->Exception = std::move(Exception);

2436 + m_pendingCreate.reset();

2437 + PendingCreate->Completed.SetEvent();

2438 +}

2439 +

2440 +void WSLCSession::WaitForConflictingCreateToComplete(std::unique_lock<std::mutex>& ContainersLock)

2441 +{

2442 + while (m_pendingCreate)

2443 + {

2444 + auto pendingCreate = m_pendingCreate;

2445 + ContainersLock.unlock();

2446 +

2447 + WaitForPendingCreateCompletion(pendingCreate);

2448 +

2449 + ContainersLock.lock();

2450 + }

2451 +}

2452 +

2453 +void WSLCSession::OnContainerCreated(const std::string& ContainerId, std::int64_t Time) noexcept

2454 +try

2455 +{

2456 + std::lock_guard containersLock{m_containersLock};

2457 +

2458 + // Containers created behind our back (BuildKit, for instance) have no pending create to match.

2459 + if (!m_pendingCreate || m_pendingCreate->Container->ID() != ContainerId)

2460 + {

2461 + return;

2462 + }

2463 +

2464 + auto pendingCreate = m_pendingCreate;

2465 + std::exception_ptr exception;

2466 +

2467 + try

2468 + {

2469 + // Key the map by Docker's container ID, which is set in the WSLCContainerImpl constructor and stable for its lifetime.

2470 + WI_VERIFY(m_containers.emplace(ContainerId, pendingCreate->Container).second);

2471 + pendingCreate->Container->RecordEvent("create", Time);

2472 + }

2473 + catch (...)

2474 + {

2475 + // Hand the failure to the waiting create rather than letting it return a container the session isn't tracking.

2476 + exception = std::current_exception();

2477 + }

2478 +

2479 + CompletePendingCreate(pendingCreate, std::move(exception));

2480 +}

2481 +CATCH_LOG()

2482 +

2483 HRESULT WSLCSession::OpenContainer(LPCSTR Id, IWSLCContainer** Container)

2484 try

2485 {

@@ -3342,6 +3450,9 @@ try

3450 if (!m_sessionTerminatingEvent.is_signaled())

3451 {

3452 m_sessionTerminatingEvent.SetEvent();

3453 +

3454 + // Wake any readers parked in an event stream so they abort instead of waiting forever.

3455 + m_eventStore.OnSessionTerminating();

3456 }

3457

3458 // Cancel any pending IO on user-provided handles to unblock operations

@@ -3907,6 +4018,23 @@ try

4018 }

4019 CATCH_RETURN();

4020

4021 +HRESULT WSLCSession::GetEvents(LONGLONG SinceTime, LONGLONG UntilTime, const WSLCFilter* Filters, ULONG FiltersCount, IWSLCEventStream** Stream)

4022 +try

4023 +{

4024 + WSLCExecutionContext context(this);

4025 +

4026 + RETURN_HR_IF_NULL(E_POINTER, Stream);

4027 +

4028 + *Stream = nullptr;

4029 +

4030 + auto filters = wsl::windows::common::wslutil::ParseKeyMultiValuePairs(Filters, FiltersCount);

4031 + auto stream = m_eventStore.CreateStream(Microsoft::WRL::ComPtr<WSLCSession>{this}, SinceTime, UntilTime, std::move(filters));

4032 +

4033 + *Stream = stream.Detach();

4034 + return S_OK;

4035 +}

4036 +CATCH_RETURN();

4037 +

4038 void WSLCSession::RecoverExistingContainers()

4039 {

4040 WI_ASSERT(m_runtime.HasDocker());

@@ -3940,7 +4068,7 @@ void WSLCSession::RecoverExistingContainers()

4068 try

4069 {

4070 auto container = WSLCContainerImpl::Open(

3943 - dockerContainer, *this, m_runtime, m_pluginNotifier.get(), std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1));

4071 + dockerContainer, *this, m_runtime, m_pluginNotifier.get(), std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1), m_eventStore);

4072

4073 auto [it, inserted] = m_containers.emplace(container->ID(), std::move(container));

4074 WI_ASSERT(inserted);

src/windows/wslcsession/WSLCSession.h

+42 -5

@@ -24,6 +24,7 @@ Abstract:

24 #include "WSLCNetworkMetadata.h"

25 #include "DockerEventTracker.h"

26 #include "DockerHTTPClient.h"

27 +#include "EventStore.h"

28 #include "IORelay.h"

29 #include <atomic>

30 #include <list>

@@ -114,6 +115,15 @@ public:

115 IFACEMETHOD(GetTerminationEvent)(_Out_ HANDLE* Event) override;

116 IFACEMETHOD(GetTerminationReason)(_Out_ WSLCVirtualMachineTerminationReason* Reason, _Out_ LPWSTR* Details) override;

117

118 + // Event streaming. Opens a stream object that yields matching events one at a time via

119 + // IWSLCEventStream::GetNext.

120 + IFACEMETHOD(GetEvents)(

121 + _In_ LONGLONG SinceTime,

122 + _In_ LONGLONG UntilTime,

123 + _In_reads_opt_(FiltersCount) const WSLCFilter* Filters,

124 + _In_ ULONG FiltersCount,

125 + _Outptr_ IWSLCEventStream** Stream) override;

126 +

127 // Image management.

128 IFACEMETHOD(PullImage)(

129 _In_ LPCSTR Image,

@@ -262,11 +272,6 @@ public:

272 UserCOMCallback RegisterUserCOMCallback();

273 void UnregisterUserCOMCallback(DWORD ThreadId);

274

265 - HANDLE SessionTerminatingEvent() const noexcept

266 - {

267 - return m_sessionTerminatingEvent.get();

268 - }

269 -

275 ULONG Id() const noexcept

276 {

277 return m_id;

@@ -319,6 +324,30 @@ private:

324

325 void CreateContainerImpl(const WSLCContainerOptions* Options, IWSLCContainer** Container);

326

327 + // A create RPC hands the container off to the Docker event stream thread here and waits, so the

328 + // container is committed and its create event recorded from that thread. Recording it from the RPC

329 + // thread instead would let another container's event be recorded first and regress the event

330 + // stream's timestamps.

331 + struct PendingContainerCreate

332 + {

333 + wil::unique_event Completed{wil::EventOptions::ManualReset};

334 + std::shared_ptr<WSLCContainerImpl> Container;

335 + std::exception_ptr Exception;

336 + };

337 +

338 + __requires_lock_held(m_containersLock) std::shared_ptr<PendingContainerCreate> StartPendingCreate(std::shared_ptr<WSLCContainerImpl> Container);

339 +

340 + __requires_lock_held(m_containersLock) void CompletePendingCreate(

341 + const std::shared_ptr<PendingContainerCreate>& PendingCreate, std::exception_ptr Exception) noexcept;

342 +

343 + void WaitForPendingCreateCompletion(const std::shared_ptr<PendingContainerCreate>& PendingCreate);

344 +

345 + // Returns with the lock held once no create is in flight. Only one fits: the slot is unkeyed,

346 + // because the container ID isn't known until Docker assigns it.

347 + void WaitForConflictingCreateToComplete(std::unique_lock<std::mutex>& ContainersLock);

348 +

349 + void OnContainerCreated(const std::string& ContainerId, std::int64_t Time) noexcept;

350 +

351 void ConfigureStorage(const WSLCSessionInitSettings& Settings, PSID UserSid);

352

353 void Ext4Format(const std::string& Device);

@@ -388,6 +417,14 @@ private:

417

418 wil::com_ptr<IWSLCPluginNotifier> m_pluginNotifier;

419

420 + // Bounded in-memory ring of recent lifecycle events, shared by all event-stream subscribers.

421 + EventStore m_eventStore;

422 +

423 + __guarded_by(m_containersLock) std::shared_ptr<PendingContainerCreate> m_pendingCreate;

424 +

425 + // N.B. Declared after everything OnContainerCreated() touches so the callback is unregistered first.

426 + DockerEventTracker::EventTrackingReference m_containerEventTracking;

427 +

428 // User-provided handles that the session is currently doing IO on.

429 std::mutex m_userHandlesLock;

430 __guarded_by(m_userHandlesLock) std::vector<HANDLE> m_userHandles;

test/windows/WSLCTests.cpp

+289

@@ -27,6 +27,7 @@ Abstract:

27 #include "WSLCSessionDefaults.h"

28 #include <nlohmann/json.hpp>

30 +using namespace std::chrono;

31 using namespace std::literals::chrono_literals;

32 using namespace wsl::windows::common::registry;

33 using wsl::windows::common::ClientRunningWSLCProcess;

@@ -6857,6 +6858,294 @@ class WSLCTests

6858 }

6859 }

6860

6861 + WSLC_TEST_METHOD(EventStream)

6862 + {

6863 + constexpr auto c_containerName = "wslc-test-events";

6864 + constexpr auto c_imageName = "debian:latest";

6865 + constexpr auto c_labelKey = "event-label";

6866 + constexpr auto c_labelValue = "event-value";

6867 + const auto expectedExitCode = std::to_string(128 + WSLCSignalSIGKILL);

6868 +

6869 + auto now = [] { return duration_cast<seconds>(system_clock::now().time_since_epoch()).count(); };

6870 +

6871 + // Drains a bounded event stream to completion (GetNext returns WSLC_E_EVENT_STREAM_FINISHED

6872 + // once the until-time has passed and the backlog is exhausted), parsing each event's JSON.

6873 + auto drain = [](IWSLCEventStream* stream) {

6874 + std::vector<wsl::windows::common::wslc_schema::Event> events;

6875 +

6876 + wil::unique_cotaskmem_ansistring eventJson;

6877 + HRESULT result;

6878 + while (SUCCEEDED(result = stream->GetNext(&eventJson)))

6879 + {

6880 + events.push_back(wsl::shared::FromJson<wsl::windows::common::wslc_schema::Event>(eventJson.get()));

6881 + }

6882 +

6883 + VERIFY_ARE_EQUAL(WSLC_E_EVENT_STREAM_FINISHED, result);

6884 + return events;

6885 + };

6886 +

6887 + // Verifies the given events match the expected actions in order for a given actor.

6888 + auto verifyEvents = [&](const std::vector<wsl::windows::common::wslc_schema::Event>& events,

6889 + const std::string& actorId,

6890 + const std::vector<std::string>& expectedActions) {

6891 + VERIFY_ARE_EQUAL(events.size(), expectedActions.size());

6892 +

6893 + for (size_t i = 0; i < expectedActions.size(); ++i)

6894 + {

6895 + const auto& action = expectedActions[i];

6896 + const auto& event = events[i];

6897 +

6898 + VERIFY_ARE_EQUAL(action, event.Action);

6899 + VERIFY_ARE_EQUAL(actorId, event.Actor.ID);

6900 + VERIFY_ARE_EQUAL(c_containerName, event.Actor.Attributes.at("name"));

6901 + VERIFY_ARE_EQUAL(c_imageName, event.Actor.Attributes.at("image"));

6902 + VERIFY_ARE_EQUAL(c_labelValue, event.Actor.Attributes.at(c_labelKey));

6903 + VERIFY_IS_FALSE(event.Actor.Attributes.contains("com.microsoft.wsl.container.metadata"));

6904 +

6905 + if (action == "stop")

6906 + {

6907 + VERIFY_ARE_EQUAL(expectedExitCode, event.Actor.Attributes.at("exitCode"));

6908 + }

6909 + else

6910 + {

6911 + VERIFY_IS_FALSE(event.Actor.Attributes.contains("exitCode"));

6912 + }

6913 + }

6914 + };

6915 +

6916 + // Run a container through its create/start/kill/stop lifecycle inside a bounded time window.

6917 + const LONGLONG since = now();

6918 + std::string id;

6919 + {

6920 + WSLCContainerLauncher launcher(c_imageName, c_containerName, {"sleep", "99999"});

6921 + launcher.AddLabel(c_labelKey, c_labelValue);

6922 + auto container = launcher.Launch(*m_defaultSession);

6923 + id = container.Id();

6924 +

6925 + VERIFY_ARE_EQUAL(container.State(), WslcContainerStateRunning);

6926 +

6927 + // Kill (rather than Stop) so Docker emits a 'kill' event ahead of the 'die' that stops it.

6928 + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGKILL));

6929 + VERIFY_ARE_EQUAL(container.State(), WslcContainerStateExited);

6930 + }

6931 +

6932 + const LONGLONG until = now() + 1;

6933 + std::vector<wsl::windows::common::wslc_schema::Event> lifecycleEvents;

6934 +

6935 + // The container's create, start, kill, stop, then destroy events are reported in order, each carrying

6936 + // the container's 64-hex id as the actor.

6937 + {

6938 + WSLCFilter filter{"container", id.c_str()};

6939 + wil::com_ptr<IWSLCEventStream> stream;

6940 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, &filter, 1, &stream));

6941 +

6942 + lifecycleEvents = drain(stream.get());

6943 + verifyEvents(lifecycleEvents, id, {"create", "start", "kill", "stop", "destroy"});

6944 +

6945 + // The whole lifecycle falls inside the requested window.

6946 + VERIFY_IS_TRUE(lifecycleEvents[0].time >= since);

6947 + VERIFY_IS_TRUE(lifecycleEvents[4].time < until);

6948 + }

6949 +

6950 + // Each lifecycle action is independently selectable: an 'event=<action>' filter, AND'd with

6951 + // the container filter, returns exactly that one event out of the five recorded above.

6952 + auto verifyEventFilter = [&](const char* action) {

6953 + WSLCFilter filters[]{{"container", id.c_str()}, {"event", action}};

6954 + wil::com_ptr<IWSLCEventStream> stream;

6955 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, filters, ARRAYSIZE(filters), &stream));

6956 +

6957 + verifyEvents(drain(stream.get()), id, {action});

6958 + };

6959 +

6960 + verifyEventFilter("create");

6961 + verifyEventFilter("start");

6962 + verifyEventFilter("kill");

6963 + verifyEventFilter("stop");

6964 + verifyEventFilter("destroy");

6965 +

6966 + // Values sharing a filter key are OR'd.

6967 + {

6968 + WSLCFilter filters[]{{"container", id.c_str()}, {"event", "create"}, {"event", "destroy"}};

6969 + wil::com_ptr<IWSLCEventStream> stream;

6970 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, filters, ARRAYSIZE(filters), &stream));

6971 +

6972 + verifyEvents(drain(stream.get()), id, {"create", "destroy"});

6973 + }

6974 +

6975 + // Image events are not recorded yet, so a 'type=image' filter excludes the container's

6976 + // events and leaves the stream empty.

6977 + {

6978 + WSLCFilter filter{"type", "image"};

6979 + wil::com_ptr<IWSLCEventStream> stream;

6980 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, &filter, 1, &stream));

6981 +

6982 + VERIFY_IS_TRUE(drain(stream.get()).empty());

6983 + }

6984 +

6985 + // An unmatched container id yields an empty stream, and GetNext validates its out-pointer.

6986 + {

6987 + WSLCFilter filter{"container", "0000000000000000000000000000000000000000000000000000000000000000"};

6988 + wil::com_ptr<IWSLCEventStream> stream;

6989 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, &filter, 1, &stream));

6990 +

6991 + VERIFY_IS_TRUE(drain(stream.get()).empty());

6992 + }

6993 +

6994 + // A since-time later than a non-zero until-time describes a backwards window and is rejected.

6995 + {

6996 + wil::com_ptr<IWSLCEventStream> stream;

6997 + VERIFY_ARE_EQUAL(E_INVALIDARG, m_defaultSession->GetEvents(since + 1, since, nullptr, 0, &stream));

6998 + ValidateCOMErrorMessage(wsl::shared::Localization::MessageWslcEventsInvalidTimeWindow(since + 1, since));

6999 + }

7000 + }

7001 +

7002 + WSLC_TEST_METHOD(EventStreamReportsLostEvents)

7003 + {

7004 + // One more than the store's ring capacity, so the reader's next slot is guaranteed evicted.

7005 + constexpr size_t c_signalsToEvictReader = 257;

7006 +

7007 + WSLCContainerLauncher launcher("debian:latest", "wslc-test-event-stream-overrun", {"sleep", "99999"});

7008 + auto container = launcher.Launch(*m_defaultSession);

7009 + const auto id = container.Id();

7010 +

7011 + WSLCFilter filter{"container", id.c_str()};

7012 + wil::com_ptr<IWSLCEventStream> stream;

7013 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(0, 0, &filter, 1, &stream));

7014 +

7015 + // Read one event to place the reader's cursor inside the ring.

7016 + wil::unique_cotaskmem_ansistring eventJson;

7017 + VERIFY_SUCCEEDED(stream->GetNext(&eventJson));

7018 + const auto firstEvent = wsl::shared::FromJson<wsl::windows::common::wslc_schema::Event>(eventJson.get());

7019 + VERIFY_ARE_EQUAL("create", firstEvent.Action);

7020 + VERIFY_ARE_EQUAL(id, firstEvent.Actor.ID);

7021 +

7022 + // Docker emits a 'kill' event per signal. SIGWINCH is ignored by an unhandling init process, so the

7023 + // container keeps running and each signal costs only one event.

7024 + for (size_t i = 0; i < c_signalsToEvictReader; ++i)

7025 + {

7026 + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGWINCH));

7027 + }

7028 +

7029 + // Stopping waits for the 'die' event, which Docker delivers after every preceding 'kill'. Without this

7030 + // barrier the reader could be checked before the ring has overrun it.

7031 + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGKILL));

7032 + VERIFY_ARE_EQUAL(container.State(), WslcContainerStateExited);

7033 +

7034 + VERIFY_ARE_EQUAL(WSLC_E_EVENTS_LOST, stream->GetNext(&eventJson));

7035 +

7036 + // Reporting the gap resyncs the reader, so it resumes from the oldest event still buffered.

7037 + VERIFY_SUCCEEDED(stream->GetNext(&eventJson));

7038 + const auto resumedEvent = wsl::shared::FromJson<wsl::windows::common::wslc_schema::Event>(eventJson.get());

7039 + VERIFY_ARE_EQUAL("kill", resumedEvent.Action);

7040 + VERIFY_ARE_EQUAL(id, resumedEvent.Actor.ID);

7041 + }

7042 +

7043 + WSLC_TEST_METHOD(EventStreamSerializesConcurrentReaders)

7044 + {

7045 + constexpr auto c_containerName = "wslc-test-concurrent-event-readers";

7046 +

7047 + WSLCContainerLauncher launcher("debian:latest", c_containerName, {"sleep", "99999"});

7048 + auto container = launcher.Launch(*m_defaultSession);

7049 + const auto id = container.Id();

7050 +

7051 + WSLCFilter filters[]{{"container", id.c_str()}, {"event", "kill"}};

7052 + wil::com_ptr<IWSLCEventStream> stream;

7053 +

7054 + // The window doubles as a hang guard, so it must comfortably outlast the waits below.

7055 + const LONGLONG until = duration_cast<seconds>(system_clock::now().time_since_epoch()).count() + 120;

7056 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(0, until, filters, ARRAYSIZE(filters), &stream));

7057 +

7058 + wil::unique_cotaskmem_ansistring firstEventJson;

7059 + wil::unique_cotaskmem_ansistring secondEventJson;

7060 + HRESULT firstResult{};

7061 + HRESULT secondResult{};

7062 + wil::unique_event firstReaderStarted{wil::EventOptions::ManualReset};

7063 + wil::unique_event secondReaderStarted{wil::EventOptions::ManualReset};

7064 + std::thread firstReader;

7065 + std::thread secondReader;

7066 +

7067 + auto cleanup = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&]() {

7068 + LOG_IF_FAILED(container.Get().Kill(WSLCSignalSIGWINCH));

7069 + LOG_IF_FAILED(container.Get().Kill(WSLCSignalSIGWINCH));

7070 +

7071 + if (firstReader.joinable())

7072 + {

7073 + firstReader.join();

7074 + }

7075 +

7076 + if (secondReader.joinable())

7077 + {

7078 + secondReader.join();

7079 + }

7080 + });

7081 +

7082 + firstReader = std::thread([&]() {

7083 + firstReaderStarted.SetEvent();

7084 + firstResult = stream->GetNext(&firstEventJson);

7085 + });

7086 + VERIFY_IS_TRUE(firstReaderStarted.wait(30 * 1000));

7087 + VERIFY_ARE_EQUAL(WAIT_TIMEOUT, WaitForSingleObject(firstReader.native_handle(), 100));

7088 +

7089 + secondReader = std::thread([&]() {

7090 + secondReaderStarted.SetEvent();

7091 + secondResult = stream->GetNext(&secondEventJson);

7092 + });

7093 + VERIFY_IS_TRUE(secondReaderStarted.wait(30 * 1000));

7094 + VERIFY_ARE_EQUAL(WAIT_TIMEOUT, WaitForSingleObject(secondReader.native_handle(), 100));

7095 +

7096 + HANDLE readers[]{firstReader.native_handle(), secondReader.native_handle()};

7097 + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGWINCH));

7098 +

7099 + const DWORD completedReader = WaitForMultipleObjects(ARRAYSIZE(readers), readers, FALSE, 30 * 1000);

7100 + VERIFY_IS_TRUE(completedReader == WAIT_OBJECT_0 || completedReader == WAIT_OBJECT_0 + 1);

7101 +

7102 + // One event completes exactly one call; the other stays serialized until another event arrives.

7103 + const DWORD pendingReader = completedReader == WAIT_OBJECT_0 ? 1 : 0;

7104 + VERIFY_ARE_EQUAL(WAIT_TIMEOUT, WaitForSingleObject(readers[pendingReader], 100));

7105 +

7106 + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGWINCH));

7107 + VERIFY_ARE_EQUAL(WAIT_OBJECT_0, WaitForMultipleObjects(ARRAYSIZE(readers), readers, TRUE, 30 * 1000));

7108 +

7109 + firstReader.join();

7110 + secondReader.join();

7111 + cleanup.release();

7112 +

7113 + VERIFY_SUCCEEDED(firstResult);

7114 + VERIFY_SUCCEEDED(secondResult);

7115 +

7116 + const auto firstEvent = wsl::shared::FromJson<wsl::windows::common::wslc_schema::Event>(firstEventJson.get());

7117 + const auto secondEvent = wsl::shared::FromJson<wsl::windows::common::wslc_schema::Event>(secondEventJson.get());

7118 + VERIFY_ARE_EQUAL("kill", firstEvent.Action);

7119 + VERIFY_ARE_EQUAL(id, firstEvent.Actor.ID);

7120 + VERIFY_ARE_EQUAL("kill", secondEvent.Action);

7121 + VERIFY_ARE_EQUAL(id, secondEvent.Actor.ID);

7122 + }

7123 +

7124 + WSLC_TEST_METHOD(EventStreamSessionTerminationAbortsReader)

7125 + {

7126 + WSLCFilter filter{"type", "container"};

7127 + const LONGLONG since = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();

7128 + wil::com_ptr<IWSLCEventStream> stream;

7129 + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, 0, &filter, 1, &stream));

7130 +

7131 + std::promise<HRESULT> getNextResult;

7132 + std::thread readerThread([&]() {

7133 + wil::unique_cotaskmem_ansistring eventJson;

7134 + getNextResult.set_value(stream->GetNext(&eventJson));

7135 + });

7136 + auto threadCleanup = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&]() { readerThread.join(); });

7137 +

7138 + auto future = getNextResult.get_future();

7139 +

7140 + VERIFY_SUCCEEDED(m_defaultSession->Terminate());

7141 + auto restore = ResetTestSession();

7142 +

7143 + // Termination wakes the parked reader; it must finish quickly and report E_ABORT.

7144 + FAIL_FAST_IF_MSG(

7145 + future.wait_for(10s) != std::future_status::ready, "event stream reader did not abort after session termination");

7146 + VERIFY_ARE_EQUAL(E_ABORT, future.get());

7147 + }

7148 +

7149 WSLC_TEST_METHOD(OpenContainer)

7150 {

7151 auto expectOpen = [&](const char* Id, HRESULT expectedResult = S_OK) {