@@ -10,7 +10,9 @@ const {
   PromisePrototypeThen,
   PromiseResolve,
   PromiseWithResolvers,
+  SafePromiseRace,
   SafeSet,
+  Symbol,
   SymbolAsyncIterator,
   SymbolDispose,
   SymbolIterator,
@@ -66,6 +68,9 @@ const {
 // Async Share Implementation
 // =============================================================================
 
+const kNoShareError = Symbol('kNoShareError');
+const kShareCancelled = Symbol('kShareCancelled');
+
 class ShareImpl {
   #source;
   #options;
@@ -78,6 +83,9 @@ class ShareImpl {
   #cancelled = false;
   #pulling = false;
   #pullWaiters = [];
+  #cancelPromise;
+  #resolveCancel;
+  #cancelError = kNoShareError;
   #cachedMinCursor = 0;
   #cachedMinCursorConsumers = 0;
   /** Cumulative byte size of buffered entries */
@@ -86,6 +94,9 @@ class ShareImpl {
   constructor(source, options) {
     this.#source = source;
     this.#options = options;
+    const { promise, resolve } = PromiseWithResolvers();
+    this.#cancelPromise = promise;
+    this.#resolveCancel = resolve;
   }
 
   get consumerCount() {
@@ -124,6 +135,7 @@ class ShareImpl {
       resolve: null,
       reject: null,
       detached: false,
+      error: kNoShareError,
       pendingNext: PromiseResolve(),
     };
 
@@ -142,25 +154,21 @@ class ShareImpl {
       __proto__: null,
       [SymbolAsyncIterator]() {
         const getNext = async () => {
-          if (self.#sourceError !== undefined) {
-            state.detached = true;
-            self.#consumers.delete(state);
-            throw self.#sourceError;
-          }
-
           // Loop until we get data, source is exhausted, or
           // consumer is detached. Multiple consumers may be woken
           // after a single pull - those that find no data at their
           // cursor must re-pull rather than terminating prematurely.
           for (;;) {
             if (state.detached) {
-              if (self.#sourceError !== undefined) throw self.#sourceError;
+              if (state.error !== kNoShareError) throw state.error;
               return { __proto__: null, done: true, value: undefined };
             }
 
             if (self.#cancelled) {
               state.detached = true;
+              state.error = self.#cancelError;
               self.#deleteConsumer(state);
+              if (state.error !== kNoShareError) throw state.error;
               return { __proto__: null, done: true, value: undefined };
             }
 
@@ -180,16 +188,20 @@ class ShareImpl {
             if (self.#sourceExhausted) {
               state.detached = true;
               self.#deleteConsumer(state);
-              if (self.#sourceError !== undefined) throw self.#sourceError;
+              if (self.#sourceError !== undefined) {
+                state.error = self.#sourceError;
+                throw state.error;
+              }
               return { __proto__: null, done: true, value: undefined };
             }
 
             // Need to pull from source - check buffer limit
             const shouldBuffer = await self.#waitForBufferSpace();
             if (shouldBuffer === null) {
               state.detached = true;
+              state.error = self.#cancelError;
               self.#deleteConsumer(state);
-              if (self.#sourceError !== undefined) throw self.#sourceError;
+              if (state.error !== kNoShareError) throw state.error;
               return { __proto__: null, done: true, value: undefined };
             }
 
@@ -241,14 +253,23 @@ class ShareImpl {
     this.#cancelled = true;
 
     if (reason !== undefined) {
-      this.#sourceError = reason;
+      this.#cancelError = reason;
     }
 
+    this.#resolveCancel(kShareCancelled);
+    this.#resolveCancel = null;
+
     if (this.#sourceIterator?.return) {
-      PromisePrototypeThen(this.#sourceIterator.return(), undefined, () => {});
+      try {
+        PromisePrototypeThen(
+          PromiseResolve(this.#sourceIterator.return()), undefined, () => {});
+      } catch {
+        // Cancellation has precedence over source cleanup errors.
+      }
     }
 
     for (const consumer of this.#consumers) {
+      consumer.error = this.#cancelError;
       if (consumer.resolve) {
         if (reason !== undefined) {
           consumer.reject?.(reason);
@@ -261,6 +282,8 @@ class ShareImpl {
       consumer.detached = true;
     }
     this.#consumers.clear();
+    this.#buffer.clear();
+    this.#bufferedBytes = 0;
 
     for (let i = 0; i < this.#pullWaiters.length; i++) {
       this.#pullWaiters[i]();
@@ -364,7 +387,12 @@ class ShareImpl {
           }
         }
 
-        const result = await this.#sourceIterator.next();
+        const result = await SafePromiseRace([
+          this.#sourceIterator.next(),
+          this.#cancelPromise,
+        ]);
+
+        if (this.#cancelled || result === kShareCancelled) return;
 
         if (result.done) {
           this.#sourceExhausted = true;
 
@@ -143,6 +143,70 @@ async function testShareCancelWithFalsyReason() {
   }
 }
 
+async function testShareCancelWhileSourcePullPending() {
+  const noReason = { __proto__: null };
+
+  for (const reason of [noReason, 0]) {
+    const sourceStarted = Promise.withResolvers();
+    const sourceNext = Promise.withResolvers();
+    let nextCalls = 0;
+    let returnCalls = 0;
+    const source = {
+      __proto__: null,
+      [Symbol.asyncIterator]() {
+        return {
+          __proto__: null,
+          next() {
+            nextCalls++;
+            sourceStarted.resolve();
+            return sourceNext.promise;
+          },
+          async return() {
+            returnCalls++;
+            return { __proto__: null, done: true, value: undefined };
+          },
+        };
+      },
+    };
+    const shared = share(source);
+    const iterator = shared.pull()[Symbol.asyncIterator]();
+    const read = iterator.next().then(
+      (value) => ({ __proto__: null, rejected: false, value }),
+      (error) => ({ __proto__: null, rejected: true, error }),
+    );
+
+    await sourceStarted.promise;
+    if (reason === noReason) {
+      shared.cancel();
+    } else {
+      shared.cancel(reason);
+    }
+
+    const timedOut = { __proto__: null };
+    const outcome = await Promise.race([
+      read,
+      new Promise((resolve) => setImmediate(resolve, timedOut)),
+    ]);
+    assert.notStrictEqual(outcome, timedOut);
+    if (reason === noReason) {
+      assert.strictEqual(outcome.rejected, false);
+      assert.strictEqual(outcome.value.done, true);
+    } else {
+      assert.strictEqual(outcome.rejected, true);
+      assert.strictEqual(outcome.error, reason);
+    }
+    assert.strictEqual(nextCalls, 1);
+
+    sourceNext.resolve({
+      __proto__: null,
+      done: false,
+      value: [Uint8Array.of(1)],
+    });
+    await new Promise(setImmediate);
+    assert.strictEqual(returnCalls, 1);
+  }
+}
+
 async function testShareAbortSignal() {
   const ac = new AbortController();
   const reason = new Error('share aborted');
@@ -272,6 +336,27 @@ async function testShareSourceError() {
   }, { message: 'share source boom' });
 }
 
+async function testShareSourceErrorFollowsBufferedData() {
+  const reason = new Error('share source boom');
+  async function* failingSource() {
+    yield [Uint8Array.of(1)];
+    throw reason;
+  }
+
+  const shared = share(failingSource());
+  const fast = shared.pull()[Symbol.asyncIterator]();
+  const slow = shared.pull()[Symbol.asyncIterator]();
+  const returned = shared.pull()[Symbol.asyncIterator]();
+  await returned.return();
+
+  assert.deepStrictEqual((await fast.next()).value, [Uint8Array.of(1)]);
+  await assert.rejects(fast.next(), (error) => error === reason);
+
+  assert.deepStrictEqual((await slow.next()).value, [Uint8Array.of(1)]);
+  await assert.rejects(slow.next(), (error) => error === reason);
+  assert.strictEqual((await returned.next()).done, true);
+}
+
 async function testShareLateJoiningConsumer() {
   // A consumer that joins after some data has been consumed should only
   // see data remaining in the buffer (not items already trimmed).
@@ -380,12 +465,14 @@ Promise.all([
   testShareCancelMidIteration(),
   testShareCancelWithReason(),
   testShareCancelWithFalsyReason(),
+  testShareCancelWhileSourcePullPending(),
   testShareAbortSignal(),
   testShareAbortSignalWhileSourcePullPending(),
   testSharePullAbortSignalRejectsPendingNext(),
   testSharePullPreAbortedSignalDoesNotAddConsumer(),
   testShareAlreadyAborted(),
   testShareSourceError(),
+  testShareSourceErrorFollowsBufferedData(),
   testShareLateJoiningConsumer(),
   testShareConsumerBreak(),
   testShareMultipleConsumersConcurrentPull(),