diff --git a/examples/src/download_actor.gleam b/examples/src/download_actor.gleam
index 0f28e9d..0b3cf00 100644
--- a/examples/src/download_actor.gleam
+++ b/examples/src/download_actor.gleam
@@ -1,17 +1,23 @@
 import crew
-import gleam/erlang/process
+import gleam/erlang/process.{type Subject}
 import gleam/int
 import gleam/io
 import gleam/otp/actor
+import gleam/string
 
 pub type DownloadMsg {
   StartDownload(url: String)
   DownloadComplete(String)
+  PoolWorkerCrashed(process.ExitReason)
   Shutdown
 }
 
 pub type State {
-  State(channel: crew.Channel(String), pending: Int)
+  State(
+    pool: process.Name(crew.PoolMsg),
+    results: Subject(Result(String, process.ExitReason)),
+    pending: Int,
+  )
 }
 
 pub fn main() {
@@ -42,15 +48,20 @@ pub fn main() {
 
 fn init(pool_name, self) {
   // open up a channel to the pool, allowing for asynchronous communication
-  let channel = crew.subscribe(pool_name)
+  let results = process.new_subject()
 
   // link the actor to the pool and subscribe to finished work
   let selector =
     process.new_selector()
     |> process.select(self)
-    |> crew.select_map(channel, DownloadComplete)
+    |> process.select_map(results, fn(result) {
+      case result {
+        Ok(result) -> DownloadComplete(result)
+        Error(down) -> PoolWorkerCrashed(down)
+      }
+    })
 
-  let state = State(channel:, pending: 0)
+  let state = State(pool: pool_name, results:, pending: 0)
 
   actor.initialised(state)
   |> actor.selecting(selector)
@@ -64,7 +75,7 @@ fn loop(state: State, msg: DownloadMsg) {
   case msg {
     StartDownload(url) -> {
       // queue a download on the worker pool using the channel.
-      crew.enqueue(state.channel, fn() { download_file(url) })
+      crew.enqueue(state.pool, state.results, fn() { download_file(url) })
       actor.continue(State(..state, pending: state.pending + 1))
     }
     DownloadComplete(result) -> {
@@ -79,6 +90,11 @@ fn loop(state: State, msg: DownloadMsg) {
       }
     }
 
+    PoolWorkerCrashed(reason) -> {
+      io.println("Worker crashed: " <> string.inspect(reason))
+      actor.stop()
+    }
+
     Shutdown -> actor.stop()
   }
 }
diff --git a/src/crew.gleam b/src/crew.gleam
index c458509..cc26599 100644
--- a/src/crew.gleam
+++ b/src/crew.gleam
@@ -337,24 +337,42 @@ fn do_work(
   timeout: Int,
   work: List(fn() -> any),
 ) -> List(any) {
-  let channel = subscribe(pool)
+  let timeout_end = system_time() + timeout
+  let assert Ok(pool_pid) = process.named(pool) as "Pool is not running"
 
-  enqueue_many(channel, work)
+  let monitor = process.monitor(pool_pid)
+  let receive = process.new_subject()
+  enqueue_many(pool, receive, work)
 
   let selector =
     process.new_selector()
-    |> select_map(channel, fn(x) { x })
+    |> process.select_specific_monitor(monitor, fn(down) {
+      let msg =
+        "Pool exited while waiting for work to complete: "
+        <> string.inspect(down)
+
+      panic as msg
+    })
+    |> process.select_map(receive, fn(result) {
+      case result {
+        Ok(value) -> value
+        Error(down) -> {
+          // demonitor here in case the panic is rescued
+          process.demonitor_process(monitor)
+          let msg = "Worker exited: " <> string.inspect(down)
+          panic as msg
+        }
+      }
+    })
 
-  let timeout_end = system_time() + timeout
   let result = receive_loop(work, selector, timeout_end, [])
 
-  // sending unsubscribe unconditionally in an attempt to handle
-  // cases more gracefully where this crash is rescued by the caller.
-  unsubscribe(channel, selector)
+  process.demonitor_process(monitor)
+  cancel(pool, receive)
 
   case result {
     Ok(value) -> value
-    Error(_) -> panic as "Pool did not complete work in the alloted time"
+    Error(_) -> panic as "Pool did not complete work in time"
   }
 }
 
@@ -379,155 +397,68 @@ fn receive_loop(
 
 // -- SUBSCRIBE / UNSUBSCRIBE -------------------------------------------------
 
-/// A channel represents a typed subscription to a worker pool for receiving
-/// work results. This allows you to send work and receive results asynchronously.
-pub opaque type Channel(a) {
-  Channel(
-    pool: Subject(PoolMsg),
-    receive: Subject(WorkResult),
-    monitor: Monitor,
-  )
-}
-
-/// Create a typed subscription channel to the worker pool for work submission.
+/// The first time `enqueue` is called from a process, the pool starts to monitor
+/// that process and cancels all ongoing work in case it goes down.
 ///
-/// This function provides a lower-level interface for submitting work to the
-/// pool. In many cases calling the `work*` function from a separate process
-/// will be easier.
+/// Sometimes it is useful to manually unsubscribe and cancel all ongoing work
+/// for a subject. Doing so will also remove the monitor added in the pool once
+/// the last `receive` subject got cancelled.
 ///
-/// Subscribing to the pool creates a reference to the current process in the pool
-/// that must be cleaned up using `unsubscribe`. The returned channel can be used
-/// with `select_map` and `enqueue` for custom message handling.
-///
-/// ## Panics
-/// - If the pool is not running
-///
-/// ## Example
-///
-/// ```gleam
-/// let channel = crew.subscribe(pool_name)
-/// crew.enqueue(channel, fn() { some_work() })
-/// // Handle results with selector...
-/// crew.unsubscribe(channel, selector)
-/// ```
-pub fn subscribe(pool: Name(PoolMsg)) -> Channel(a) {
-  let pool_subject = process.named_subject(pool)
-  let assert Ok(pool_pid) = process.subject_owner(pool_subject)
-    as "Pool is not running"
-
-  let monitor = process.monitor(pool_pid)
-
-  let receive: Subject(WorkResult) = process.new_subject()
-  actor.send(pool_subject, Subscribe(receive))
-
-  Channel(pool: pool_subject, receive:, monitor:)
-}
-
-/// Add a worker pool channel to a selector for receiving work results.
-///
-/// This allows you to receive work results as part of the larger message-handling
-/// loop. Used in conjunction with `subscribe` and `enqueue` for lower-level
-/// pool usage.
-///
-/// Work results arrive in completion order. It is your responsibility to handle
-/// the lifecycle and ordering.
-///
-/// This adds a specific monitor to the selector to handle the case where the pool
-/// goes down, so it's better to prefer few channels over many.
-///
-/// ## Panics
-/// - If a worker or the pool crashes while executing work, the selector will
-/// panic with details about the crash.
-pub fn select_map(
-  selector: Selector(msg),
-  channel: Channel(a),
-  tagger: fn(a) -> msg,
-) -> Selector(msg) {
-  selector
-  |> process.select_specific_monitor(channel.monitor, fn(down) {
-    let msg =
-      "Pool exited while waiting for work to complete: " <> string.inspect(down)
-
-    panic as msg
-  })
-  |> process.select_map(channel.receive, fn(result) {
-    case result {
-      Done(value) -> tagger(cast(value))
-      WorkerExited(reason) -> {
-        let msg = "Worker exited: " <> string.inspect(reason)
-        panic as msg
-      }
-    }
-  })
-}
-
-/// Remove a subscription channel from the pool and clean up resources.
-///
-/// This function should be called when you're done using a channel obtained
-/// from `subscribe`. It stops any in-progress work for this channel and
-/// removes the channel's handlers from the selector.
-///
-/// Any finished work that is still coming in is dropped.
-pub fn unsubscribe(
-  channel: Channel(a),
-  selector: Selector(msg),
-) -> Selector(msg) {
-  let unsubscribed = process.new_subject()
-  actor.send(channel.pool, Unsubscribe(channel.receive, unsubscribed:))
-
-  wait_for_unsubscribe(channel, unsubscribed)
-  process.demonitor_process(channel.monitor)
-
-  selector
-  |> process.deselect_specific_monitor(channel.monitor)
-  |> process.deselect(channel.receive)
-}
-
-fn wait_for_unsubscribe(channel: Channel(a), unsubscribed: Subject(b)) -> Nil {
-  let unsubscribe_selector =
-    process.new_selector()
-    |> process.select_map(unsubscribed, fn(_) { True })
-    |> process.select_specific_monitor(channel.monitor, fn(_) { True })
-    |> process.select_map(channel.receive, fn(_) { False })
-
-  drop_messages_loop(unsubscribe_selector)
-}
-
-fn drop_messages_loop(selector: Selector(Bool)) -> Nil {
-  case process.selector_receive_forever(selector) {
-    True -> Nil
-    False -> drop_messages_loop(selector)
-  }
+/// Note that finished work might still arrive on this selector after
+/// unsubsribe got called.
+pub fn cancel(
+  pool: Name(PoolMsg),
+  receive: Subject(Result(a, process.ExitReason)),
+) -> Nil {
+  actor.send(process.named_subject(pool), Cancel(cast(receive)))
 }
 
 /// Submit a single piece of work to the pool using a subscription channel.
 ///
-/// This is a lower-level function for submitting work. Results must be handled
-/// using a selector with `select_map`. It is the callers responsibility to handle
-/// timeouts and submission order. Most users should prefer the `work*` functions.
-pub fn enqueue(channel: Channel(a), work: fn() -> a) -> Nil {
-  enqueue_many(channel, [work])
+/// This is a lower-level function for submitting work.  It is the callers
+/// responsibility to handle timeouts, submission order and failures.
+/// Most users should prefer the `work*` functions.
+///
+/// The first time you `enqueue` is called from a process, the pool sets up
+/// a monitor making sure work is cancelled when the process no longer exists
+/// to receive a result. You can clean up this monitor early by using `cancel`.
+pub fn enqueue(
+  pool: Name(PoolMsg),
+  receive: Subject(Result(a, process.ExitReason)),
+  work: fn() -> a,
+) -> Nil {
+  enqueue_many(pool, receive, [work])
 }
 
 /// Submit multiple pieces of work to the pool using a subscription channel.
 ///
-/// This is a lower-level function for submitting work. Results must be handled
-/// using a selector with `select_map`. It is the callers responsibility to handle
-/// timeouts and submission order. Most users should prefer the `work*` functions.
-pub fn enqueue_many(channel: Channel(a), work: List(fn() -> a)) -> Nil {
-  actor.send(channel.pool, Enqueue(channel.receive, cast(work)))
+/// This is a lower-level function for submitting work.  It is the callers
+/// responsibility to handle timeouts, submission order and failures.
+/// Most users should prefer the `work*` functions.
+///
+/// The first time you `enqueue` is called from a process, the pool sets up
+/// a monitor making sure work is cancelled when the process no longer exists
+/// to receive a result. You can clean up this monitor early by using `cancel`.
+pub fn enqueue_many(
+  pool: Name(PoolMsg),
+  receive: Subject(Result(a, process.ExitReason)),
+  work: List(fn() -> a),
+) -> Nil {
+  actor.send(process.named_subject(pool), Enqueue(cast(receive), cast(work)))
 }
 
 // -- POOL --------------------------------------------------------------------
 
+type Receiver =
+  Subject(Result(Any, process.ExitReason))
+
 pub opaque type PoolMsg {
   WorkerStarted(pid: Pid, send: Subject(Work))
   WorkerIdle(pid: Pid)
   MonitoredProcessExited(reason: process.Down)
   //
-  Subscribe(receive: Subject(WorkResult))
-  Unsubscribe(receive: Subject(WorkResult), unsubscribed: Subject(Nil))
-  Enqueue(receive: Subject(WorkResult), work: List(fn() -> Any))
+  Enqueue(receive: Receiver, work: List(fn() -> Any))
+  Cancel(receive: Receiver)
 }
 
 type State {
@@ -535,8 +466,8 @@ type State {
     idle_workers: List(Worker),
     active_workers: Dict(Pid, ActiveWorker),
     callers: Dict(Pid, Caller),
-    requests: Dict(Subject(WorkResult), Request),
-    queue: Deque(QueueItem),
+    requests: Dict(Receiver, Request),
+    queue: Deque(Work),
   )
 }
 
@@ -544,20 +475,16 @@ type Worker {
   Worker(pid: Pid, send: Subject(Work), monitor: Monitor)
 }
 
-type QueueItem {
-  QueueItem(work: fn() -> Any, caller: Pid, receive: Subject(WorkResult))
-}
-
 type ActiveWorker {
-  ActiveWorker(worker: Worker, caller: Pid, receive: Subject(WorkResult))
+  ActiveWorker(worker: Worker, work: Work)
 }
 
 type Caller {
-  Caller(pid: Pid, monitor: Monitor, requests: Set(Subject(WorkResult)))
+  Caller(pid: Pid, monitor: Monitor, requests: Set(Receiver))
 }
 
 type Request {
-  Request(from: Pid, workers: Set(Pid), receive: Subject(WorkResult))
+  Request(from: Pid, workers: Set(Pid), receive: Receiver)
 }
 
 fn init_pool(self: Subject(PoolMsg)) {
@@ -590,25 +517,36 @@ fn pool(state: State, msg: PoolMsg) -> Next(State, PoolMsg) {
       try_dequeue_work(state, worker)
     }
 
-    MonitoredProcessExited(process.ProcessDown(pid:, monitor: _, reason:)) -> {
+    MonitoredProcessExited(process.ProcessDown(pid:, ..)) -> {
       case dict.get(state.callers, pid), dict.get(state.active_workers, pid) {
         // Caller down -
         // We will get a second Down message for when the process is caller and worker.
         Ok(_), _ -> {
-          // this sends a message back to the caller, which we already
-          // know is down. This should not matter.
-          abort_caller(state, pid, reason)
+          // we do not have an active caller, maybe we still got the unsubscribe
+          // just before?
+          use caller <- try_(dict.get(state.callers, pid), state)
+
+          let requests =
+            set.fold(caller.requests, state.requests, fn(requests, receive) {
+              use request <- try_(dict.get(state.requests, receive), requests)
+              // we kill all workers currently working on things for this caller.
+              // the WorkerStopped messages will then clean up the workers.
+              set.each(request.workers, process.kill)
+
+              dict.delete(requests, receive)
+            })
+
+          // the queue is cleaned up while consuming it if the caller is missing.
+          let callers = dict.delete(state.callers, pid)
+
+          State(..state, callers:, requests:)
         }
 
         // Active worker down
-        Error(_), Ok(ActiveWorker(caller:, ..)) -> {
-          // after one worker exits, we want to stop all workers for this caller
-          // and remove it as an active caller. The caller will no longer receive messages.
-          let active_workers = dict.delete(state.active_workers, pid)
-
-          State(..state, active_workers:)
-          |> abort_caller(caller, reason)
+        Error(_), Ok(_) -> {
+          State(..state, active_workers: dict.delete(state.active_workers, pid))
         }
+
         // idle worker down
         Error(_), Error(_) -> {
           let idle_workers =
@@ -624,7 +562,7 @@ fn pool(state: State, msg: PoolMsg) -> Next(State, PoolMsg) {
 
     WorkerIdle(pid:) -> {
       // got an idle message from a worker that is not active
-      use ActiveWorker(worker:, receive:, caller: _) <- try_(
+      use ActiveWorker(worker:, work: Work(receive:, ..)) <- try_(
         dict.get(state.active_workers, pid),
         state,
       )
@@ -649,17 +587,15 @@ fn pool(state: State, msg: PoolMsg) -> Next(State, PoolMsg) {
 
     Enqueue(work: [], ..) -> state
     Enqueue(receive:, work:) -> {
-      use request <- try_(dict.get(state.requests, receive), state)
-      enqueue_loop(state, request, work)
-    }
-
-    Subscribe(receive:) -> {
+      // caller already exited, do not queue their work.
       use pid <- try_(process.subject_owner(receive), state)
 
+      // get the request data or construct a new entry if needed.
       let request =
         dict.get(state.requests, receive)
         |> result.unwrap(Request(from: pid, workers: set.new(), receive:))
 
+      // get the caller data and insert the new request, starting to monitor if needed.
       let caller = case dict.get(state.callers, pid) {
         Ok(caller) ->
           Caller(..caller, requests: set.insert(caller.requests, receive))
@@ -670,17 +606,18 @@ fn pool(state: State, msg: PoolMsg) -> Next(State, PoolMsg) {
         }
       }
 
-      let callers = dict.insert(state.callers, pid, caller)
-      let requests = dict.insert(state.requests, receive, request)
-
-      State(..state, callers:, requests:)
+      // enqueue_loop will insert the request into the state.
+      State(..state, callers: dict.insert(state.callers, pid, caller))
+      |> enqueue_loop(request, work)
     }
 
-    Unsubscribe(receive:, unsubscribed:) -> {
+    Cancel(receive:) -> {
+      // if the receiver does not exist, we don't need to do anything
       use request <- try_(dict.get(state.requests, receive), state)
 
       let requests = dict.delete(state.requests, receive)
 
+      // cancel all running workers
       set.each(request.workers, process.kill)
 
       use caller <- try_(
@@ -690,6 +627,8 @@ fn pool(state: State, msg: PoolMsg) -> Next(State, PoolMsg) {
 
       let caller =
         Caller(..caller, requests: set.delete(caller.requests, receive))
+
+      // if this request was the last one for the caller, demonitor and remove it.
       let callers = case set.is_empty(caller.requests) {
         True -> {
           process.demonitor_process(caller.monitor)
@@ -700,8 +639,6 @@ fn pool(state: State, msg: PoolMsg) -> Next(State, PoolMsg) {
         }
       }
 
-      process.send(unsubscribed, Nil)
-
       State(..state, requests:, callers:)
     }
   }
@@ -719,9 +656,10 @@ fn enqueue_loop(
   case work, state.idle_workers {
     [work, ..rest], [worker, ..idle_workers] -> {
       // got a work item and a worker - it's a match!
-      process.send(worker.send, Work(work:, receive:))
+      let work = Work(work:, caller:, receive:)
+      process.send(worker.send, work)
 
-      let active_worker = ActiveWorker(worker:, caller:, receive:)
+      let active_worker = ActiveWorker(worker:, work:)
       let active_workers =
         dict.insert(state.active_workers, worker.pid, active_worker)
 
@@ -734,7 +672,7 @@ fn enqueue_loop(
     _, _ -> {
       let queue =
         list.fold(work, state.queue, fn(queue, work) {
-          deque.push_back(queue, QueueItem(work:, caller:, receive:))
+          deque.push_back(queue, Work(work:, caller:, receive:))
         })
 
       let requests = dict.insert(state.requests, receive, request)
@@ -745,67 +683,36 @@ fn enqueue_loop(
 }
 
 fn try_dequeue_work(state: State, worker: Worker) -> State {
-  use #(QueueItem(work:, caller:, receive:), queue) <- try(
+  use #(work, queue) <- try(
     deque.pop_front(state.queue),
     // the queue is empty, this worker becomes idle.
     fn(_) { State(..state, idle_workers: [worker, ..state.idle_workers]) },
   )
 
-  use request <- try(dict.get(state.requests, receive), fn(_) {
+  use request <- try(dict.get(state.requests, work.receive), fn(_) {
     // this request unsubscribed
     try_dequeue_work(State(..state, queue:), worker)
   })
 
-  process.send(worker.send, Work(work:, receive:))
+  process.send(worker.send, work)
 
-  let active_worker = ActiveWorker(worker:, caller:, receive:)
+  let active_worker = ActiveWorker(worker:, work:)
   let active_workers =
     dict.insert(state.active_workers, worker.pid, active_worker)
 
   let request =
     Request(..request, workers: set.insert(request.workers, worker.pid))
-  let requests = dict.insert(state.requests, receive, request)
+  let requests = dict.insert(state.requests, work.receive, request)
 
   State(..state, active_workers:, requests:, queue:)
 }
 
-fn abort_caller(state: State, caller: Pid, reason: process.ExitReason) -> State {
-  // we do not have an active caller, so this worker was stopped in
-  // response to a caller exiting.
-  use caller <- try_(dict.get(state.callers, caller), state)
-
-  process.demonitor_process(caller.monitor)
-
-  let requests =
-    set.fold(caller.requests, state.requests, fn(requests, receive) {
-      use request <- try_(dict.get(state.requests, receive), requests)
-      // we kill all workers currently working on things for this caller.
-      // the WorkerStopped messages will then clean up the workers.
-      set.each(request.workers, process.kill)
-
-      // notify the caller that we crashed. this is the last message we send.
-      process.send(receive, WorkerExited(reason))
-
-      dict.delete(requests, receive)
-    })
-
-  // the queue is cleaned up while consuming it if the caller is missing.
-  let callers = dict.delete(state.callers, caller.pid)
-
-  State(..state, callers:, requests:)
-}
-
 // -- WORKER ------------------------------------------------------------------
 
 type Any
 
 type Work {
-  Work(work: fn() -> Any, receive: Subject(WorkResult))
-}
-
-type WorkResult {
-  Done(Any)
-  WorkerExited(process.ExitReason)
+  Work(work: fn() -> Any, caller: Pid, receive: Receiver)
 }
 
 fn worker(pool: Name(PoolMsg)) -> Nil {
@@ -818,11 +725,11 @@ fn worker(pool: Name(PoolMsg)) -> Nil {
 }
 
 fn worker_loop(pool: Subject(PoolMsg), self: Pid, subject: Subject(Work)) {
-  let Work(work:, receive:) = process.receive_forever(subject)
+  let Work(work:, receive:, caller: _) = process.receive_forever(subject)
 
   let result = work()
   process.send(pool, WorkerIdle(self))
-  process.send(receive, Done(result))
+  process.send(receive, Ok(result))
 
   worker_loop(pool, self, subject)
 }