volume-manager: consume medium_changed — the second removal trigger (S5)

The block client gains subscribeMedium / unsubscribeMedium /
decodeMediumChanged (the reserved subscribe/unsubscribe verbs plus the
MediumChanged decode), so a consumer never hand-rolls the wire format.
The volume manager subscribes to each device it adopts and consumes the
event through the new on_buffered_message seam — never the protocol
dispatch, whose op numbers collide with the manager's own hello.

A medium leaving while its device stays in the tree (a card reader, an
eject) now runs the SAME kill-retire-remount path as a pulled stick:
absent retires the volume, present re-probes it. That closes the "two
triggers, one lifecycle" the architecture specifies — device-presence
polling alone could never see a medium leave under a present device.
dropDevice unsubscribes before closing so the driver's bounded
subscriber table frees the slot; on a dead channel (a real pull) the
call fails fast, proven by volume-removal still passing.

New volume-medium-change case: eject the medium (not the device) ->
"medium absent" -> the manager unmounts. Fails against a pre-S5 manager
that never subscribed. Suite 132/132 (device-authority is the known
child-cleanup flake, green on rerun).

library/device/block/block.zig

@@ -7,12 +7,25 @@
 //! `runtime.dma.alloc`), so whole sectors move without crossing the IPC size
 //! limit — the same handoff usb-storage uses toward the controller.
 
+const std = @import("std");
 const envelope = @import("envelope");
 const ipc = @import("ipc");
 const block_protocol = @import("block-protocol");
 
 const Protocol = block_protocol.Protocol;
 
+/// The medium_changed event payload, re-exported so a consumer decodes it without
+/// reaching into the wire-format module.
+pub const MediumChanged = block_protocol.MediumChanged;
+
+/// Decode a medium_changed event from a buffered-message payload a subscriber
+/// received (a `Received.isMessage` wake). Null if the bytes are too short to be
+/// one — a caller ignores anything that is not a well-formed event.
+pub fn decodeMediumChanged(payload: []const u8) ?MediumChanged {
+    if (payload.len < envelope.prefix_size + @sizeOf(MediumChanged)) return null;
+    return std.mem.bytesToValue(MediumChanged, payload[envelope.prefix_size..][0..@sizeOf(MediumChanged)]);
+}
+
 pub const Geometry = struct { block_size: u32, block_count: u64 };
 
 pub const Device = struct {
@@ -88,6 +101,30 @@ pub const Device = struct {
         if (status.status != 0) return null;
         return reply[0..answer.len];
     }
+
+    /// Subscribe `subscriber` (an endpoint) to this device's medium_changed
+    /// events: the reserved `subscribe` verb carries the subscriber's endpoint as
+    /// the capability, and the driver then ipc.sends each medium transition to it.
+    pub fn subscribeMedium(self: Device, subscriber: ipc.Handle) bool {
+        var packet: [block_protocol.message_maximum]u8 = undefined;
+        const framed = envelope.encodeSubscribe(0, &packet) orelse return false; // interest 0: every event (block has one)
+        var reply: [block_protocol.message_maximum]u8 = undefined;
+        const answer = ipc.callCap(self.endpoint, framed, &reply, subscriber) catch return false;
+        const status = envelope.statusOf(reply[0..answer.len]) orelse return false;
+        return status.status == 0;
+    }
+
+    /// Unsubscribe from this device's medium_changed events. Call before closing
+    /// the channel so the driver's bounded subscriber table frees the slot rather
+    /// than holding a dead endpoint until an exit sweep notices.
+    pub fn unsubscribeMedium(self: Device) bool {
+        var packet: [block_protocol.message_maximum]u8 = undefined;
+        const framed = envelope.encodeUnsubscribe(&packet) orelse return false;
+        var reply: [block_protocol.message_maximum]u8 = undefined;
+        const answer = ipc.callCap(self.endpoint, framed, &reply, null) catch return false;
+        const status = envelope.statusOf(reply[0..answer.len]) orelse return false;
+        return status.status == 0;
+    }
 };
 
 // There is deliberately no open-by-name here: `block` is not a registry name.

system/services/volume-manager/volume-manager.zig

@@ -306,6 +306,12 @@ fn bringUpVolume() bool {
         return false;
     };
     dev.* = .{ .used = true, .device_id = opened.device_id, .channel = opened.device };
+    // Consume this device's medium_changed events (the second of the removal
+    // lifecycle's two triggers: the device stays in the tree while its medium
+    // leaves — a card reader, an eject). Best effort: a provider that never
+    // publishes the event simply never wakes us, and device-pull is still caught
+    // by the presence poll.
+    _ = opened.device.subscribeMedium(service_endpoint);
     const device = opened.device;
     // Attach the read buffer to THIS device (a no-op without an enforcing IOMMU).
     // The handle is kept, not closed, so it can be re-attached after a replug. A
@@ -370,6 +376,10 @@ fn bringUpVolume() bool {
 /// Close a device's channel and free its slot. No volumes are touched (the caller
 /// ensures none remain, or there never were any).
 fn dropDevice(dev: *StorageDevice) void {
+    // Free the driver's subscriber slot before the channel closes. On a still-live
+    // channel (a medium eject) this frees the slot; on a dead one (a device pull)
+    // the call fails fast and the exit sweep frees it anyway.
+    _ = dev.channel.unsubscribeMedium();
     _ = ipc.close(dev.channel.endpoint);
     dev.* = .{};
 }
@@ -533,6 +543,37 @@ fn onNotification(badge: u64) void {
     }
 }
 
+var last_medium_change: u32 = 0;
+
+fn anyVolumeOn(device_id: u64) bool {
+    for (&volumes) |*v| if (v.used and v.device_id == device_id) return true;
+    return false;
+}
+
+/// A storage device published `medium_changed` — the second removal trigger: the
+/// device stays in the tree while its medium leaves or returns (a card reader, an
+/// eject). This arrives as a buffered async message, NOT a protocol request, so it
+/// never reaches `Serve.dispatch` (its event op number collides with the manager's
+/// own `hello`); it is decoded here by hand. Single-volume scope: the event names
+/// no device, so `absent` retires every adopted device (its volumes unmount and
+/// the poll re-adopts the still-present device with its now-empty medium), and
+/// `present` frees any empty adopted device so the poll re-probes and remounts it.
+fn onMediumEvent(payload: []const u8) void {
+    const event = block.decodeMediumChanged(payload) orelse return;
+    if (event.change_count == last_medium_change) return; // a coalesced or re-delivered edge
+    last_medium_change = event.change_count;
+    if (event.present == 0) {
+        std.log.info("medium left a storage device; unmounting its volume(s)", .{});
+        for (&devices) |*dev| {
+            if (dev.used) removeDevice(dev);
+        }
+    } else {
+        for (&devices) |*dev| {
+            if (dev.used and !anyVolumeOn(dev.device_id)) removeDevice(dev);
+        }
+    }
+}
+
 pub fn main(init: process.Init) void {
     _ = init;
     service.run(volume_manager_protocol.message_maximum, .{
@@ -540,5 +581,6 @@ pub fn main(init: process.Init) void {
         .init = initialise,
         .on_message = onMessage,
         .on_notification = onNotification,
+        .on_buffered_message = onMediumEvent,
     });
 }

test/qemu_test.py

@@ -796,6 +796,25 @@ CASES = [
      "expect": r"(?s)fat: mounted /volumes/fat-12345678"
                r"[\s\S]*volume-manager: storage for volume \d+ removed; unmounting",
      "fail":   r"DANOS-TEST-RESULT: FAIL"},
+    # S5 medium_changed: the SECOND removal trigger. QMP-eject the MEDIUM (the
+    # block backend, not the device) — the usb-storage device stays in the tree,
+    # but its TEST UNIT READY poll reports not-ready and publishes medium_changed
+    # (absent). The volume manager, now a subscriber, runs the same unmount path as
+    # a device pull. Discrimination: before S5 the manager never subscribed, so the
+    # event reached no one and the mount persisted (device-presence polling cannot
+    # see a medium leave while the device stays). One lifecycle, two triggers.
+    {"name": "volume-medium-change",
+     "build_case": "fat-mount",
+     "smp":    4,
+     "timeout": 150,
+     "qmp_sequence": [
+         {"delay": 8, "command": "eject", "arguments": {"device": "bootusb", "force": True}},
+     ],
+     "expect": r"(?s)fat: mounted /volumes/fat-12345678"
+               r"[\s\S]*usb-storage: medium absent"
+               r"[\s\S]*volume-manager: medium left a storage device; unmounting"
+               r"[\s\S]*volume-manager: storage for volume \d+ removed; unmounting",
+     "fail":   r"DANOS-TEST-RESULT: FAIL"},
     # Volume-manager discovery + probe (V3a, docs/volume-manager-plan.md). Reuses
     # the fat-mount kernel build (the default boot now spawns the volume manager
     # from init.csv). It acquires the mass-storage block channel through the