diff --git a/src/CoreTests/Blocks/BlockErrorHandlingTests.cs b/src/CoreTests/Blocks/BlockErrorHandlingTests.cs
new file mode 100644
index 00000000..b121a8a8
--- /dev/null
+++ b/src/CoreTests/Blocks/BlockErrorHandlingTests.cs
@@ -0,0 +1,188 @@
+using JasperFx.Blocks;
+using Shouldly;
+using Xunit;
+
+namespace CoreTests.Blocks;
+
+// jasperfx#506: exceptions escaping a block's action must be observable, the error callback must
+// never be able to kill the consumer loop, and a dead (faulted) block must be observable by its
+// producers instead of silently accepting items forever
+public class BlockErrorHandlingTests
+{
+    [Fact]
+    public async Task on_error_receives_the_item_and_exception_and_the_loop_survives()
+    {
+        var failures = new List<(string, Exception)>();
+        var processed = new List<string>();
+        var secondItemProcessed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var block = new Block<string>(async (item, _) =>
+        {
+            if (item == "poison")
+            {
+                throw new DivideByZeroException("boom");
+            }
+
+            processed.Add(item);
+            if (item == "second")
+            {
+                secondItemProcessed.SetResult();
+            }
+
+            await Task.CompletedTask;
+        });
+
+        block.OnError = (item, ex) => failures.Add((item, ex));
+
+        block.Post("first");
+        block.Post("poison");
+        block.Post("second");
+
+        await secondItemProcessed.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        processed.ShouldBe(new[] { "first", "second" });
+        failures.Count.ShouldBe(1);
+        failures[0].Item1.ShouldBe("poison");
+        failures[0].Item2.ShouldBeOfType<DivideByZeroException>();
+        block.Failure.ShouldBeNull();
+
+        await block.WaitForCompletionAsync();
+    }
+
+    [Fact]
+    public async Task a_throwing_error_callback_faults_the_block_and_post_throws()
+    {
+        var faulted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var block = new Block<string>((item, _) => throw new DivideByZeroException("action"));
+
+        var callbackInvocations = 0;
+        block.OnError = (_, _) =>
+        {
+            // Only throw on the per-item invocation. The fault path re-invokes the callback with
+            // the terminal exception; letting that second call succeed lets the test observe it
+            if (Interlocked.Increment(ref callbackInvocations) == 1)
+            {
+                throw new InvalidCastException("error handling is broken too");
+            }
+
+            faulted.SetResult();
+        };
+
+        block.Post("poison");
+
+        await faulted.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        block.Failure.ShouldNotBeNull();
+        var aggregate = block.Failure.ShouldBeOfType<AggregateException>();
+        aggregate.InnerExceptions.OfType<DivideByZeroException>().Any().ShouldBeTrue();
+        aggregate.InnerExceptions.OfType<InvalidCastException>().Any().ShouldBeTrue();
+
+        // A dead consumer must be observable by producers (jasperfx#506) -- the old behavior was
+        // for Post/PostAsync to keep succeeding forever against a consumer that no longer exists
+        var thrown = Should.Throw<InvalidOperationException>(() => block.Post("after the fault"));
+        thrown.InnerException.ShouldBeSameAs(block.Failure);
+
+        await Should.ThrowAsync<InvalidOperationException>(async () => await block.PostAsync("also after"));
+    }
+
+    [Fact]
+    public async Task fault_releases_a_producer_blocked_on_a_full_bounded_channel()
+    {
+        var letTheActionFinish = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var actionStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        // Capacity 1: one item in flight blocking the consumer, one buffered, so the next
+        // PostAsync waits for capacity
+        var block = new Block<string>(1, 1, async (item, _) =>
+        {
+            actionStarted.TrySetResult();
+            await letTheActionFinish.Task;
+            if (item == "poison")
+            {
+                throw new DivideByZeroException();
+            }
+        });
+
+        block.OnError = (_, _) => throw new InvalidCastException("kill the block");
+
+        block.Post("poison");
+        await actionStarted.Task.WaitAsync(TimeSpan.FromSeconds(5));
+        block.Post("buffered");
+
+        // This producer has no capacity left and parks waiting for the consumer
+        var blockedProducer = Task.Run(async () => await block.PostAsync("parked"));
+        blockedProducer.IsCompleted.ShouldBeFalse();
+
+        // Now the action throws, the error callback throws, and the block faults. The parked
+        // producer must be released with an exception, not left hanging forever against a dead
+        // consumer (the bounded-channel hang scenario from jasperfx#506)
+        letTheActionFinish.SetResult();
+
+        await Should.ThrowAsync<Exception>(async () =>
+            await blockedProducer.WaitAsync(TimeSpan.FromSeconds(5)));
+
+        block.Failure.ShouldNotBeNull();
+    }
+
+    [Fact]
+    public async Task wait_for_completion_on_a_faulted_block_returns_quietly()
+    {
+        var faulted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var block = new Block<string>((_, _) => throw new DivideByZeroException());
+        var invocations = 0;
+        block.OnError = (_, _) =>
+        {
+            if (Interlocked.Increment(ref invocations) == 1)
+            {
+                throw new InvalidCastException();
+            }
+
+            faulted.SetResult();
+        };
+
+        block.Post("poison");
+        await faulted.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        // Teardown paths call this on shutdown; a fault must not blow up the shutdown itself
+        await block.WaitForCompletionAsync().WaitAsync(TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public async Task default_error_sink_is_not_compiled_away_in_release_builds()
+    {
+        var original = Console.Error;
+        var writer = new StringWriter();
+        Console.SetError(writer);
+
+        try
+        {
+            var processed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            var block = new Block<string>((item, _) =>
+            {
+                if (item == "poison")
+                {
+                    throw new DivideByZeroException("visible in production");
+                }
+
+                processed.SetResult();
+                return Task.CompletedTask;
+            });
+
+            // Deliberately NOT setting OnError -- this exercises the default sink, which used to
+            // be Debug.WriteLine and therefore /dev/null in release builds (jasperfx#506)
+            block.Post("poison");
+            block.Post("after");
+
+            await processed.Task.WaitAsync(TimeSpan.FromSeconds(5));
+            await block.WaitForCompletionAsync();
+
+            writer.ToString().ShouldContain("DivideByZeroException");
+        }
+        finally
+        {
+            Console.SetError(original);
+        }
+    }
+}
diff --git a/src/EventTests/Daemon/TeardownExceptionHandlingTests.cs b/src/EventTests/Daemon/TeardownExceptionHandlingTests.cs
new file mode 100644
index 00000000..ae282940
--- /dev/null
+++ b/src/EventTests/Daemon/TeardownExceptionHandlingTests.cs
@@ -0,0 +1,225 @@
+using System.Data.Common;
+using JasperFx.Events;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Microsoft.Extensions.Logging;
+using NSubstitute;
+using Shouldly;
+using Xunit;
+
+namespace EventTests.Daemon;
+
+// jasperfx#507: the teardown catch clauses used to be `catch when
+// (_cancellation.IsCancellationRequested)`, which discarded ANY exception type once the shard
+// started stopping. Only genuine cancellation side effects may be discarded silently; anything
+// else must at least be logged
+public class CancellationExceptionsTests
+{
+    [Fact]
+    public void plain_operation_canceled_is_cancellation_like()
+    {
+        CancellationExceptions.IsCancellationLike(new OperationCanceledException()).ShouldBeTrue();
+        CancellationExceptions.IsCancellationLike(new TaskCanceledException()).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void wrapped_cancellation_is_cancellation_like()
+    {
+        var wrapped = new InvalidOperationException("outer", new OperationCanceledException());
+        CancellationExceptions.IsCancellationLike(wrapped).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void aggregated_cancellation_is_cancellation_like()
+    {
+        var aggregate = new AggregateException(new OperationCanceledException(), new TaskCanceledException());
+        CancellationExceptions.IsCancellationLike(aggregate).ShouldBeTrue();
+    }
+
+    [Fact]
+    public void aggregate_containing_a_real_failure_is_not_cancellation_like()
+    {
+        var aggregate = new AggregateException(new OperationCanceledException(),
+            new FakeDbException("23514"));
+        CancellationExceptions.IsCancellationLike(aggregate).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void object_disposed_is_cancellation_like()
+    {
+        CancellationExceptions.IsCancellationLike(new ObjectDisposedException("NpgsqlConnection"))
+            .ShouldBeTrue();
+    }
+
+    [Theory]
+    [InlineData("57014", true)] // query_canceled
+    [InlineData("08006", true)] // connection_failure
+    [InlineData("08003", true)] // connection_does_not_exist
+    [InlineData("23514", false)] // check_violation -- the incident's missing-partition failure
+    [InlineData("42P01", false)] // undefined_table
+    [InlineData(null, false)]
+    public void db_exceptions_classify_by_sql_state(string? sqlState, bool expected)
+    {
+        CancellationExceptions.IsCancellationLike(new FakeDbException(sqlState)).ShouldBe(expected);
+    }
+
+    [Fact]
+    public void wrapped_db_exception_classifies_by_the_inner_sql_state()
+    {
+        // e.g. MartenCommandException wrapping the provider exception
+        var cancelledUnderneath = new InvalidOperationException("command failed", new FakeDbException("57014"));
+        CancellationExceptions.IsCancellationLike(cancelledUnderneath).ShouldBeTrue();
+
+        var schemaProblemUnderneath = new InvalidOperationException("command failed", new FakeDbException("23514"));
+        CancellationExceptions.IsCancellationLike(schemaProblemUnderneath).ShouldBeFalse();
+    }
+
+    [Fact]
+    public void arbitrary_exceptions_are_not_cancellation_like()
+    {
+        CancellationExceptions.IsCancellationLike(new DivideByZeroException()).ShouldBeFalse();
+        CancellationExceptions.IsCancellationLike(new InvalidOperationException()).ShouldBeFalse();
+    }
+}
+
+public class GroupedProjectionExecutionTeardownTests
+{
+    private readonly RecordingLogger theLogger = new();
+    private readonly IGroupedProjectionRunner theRunner = Substitute.For<IGroupedProjectionRunner>();
+    private readonly ISubscriptionAgent theAgent = Substitute.For<ISubscriptionAgent>();
+
+    public GroupedProjectionExecutionTeardownTests()
+    {
+        theAgent.Metrics.Returns(Substitute.For<ISubscriptionMetrics>());
+        theRunner.SliceBehavior.Returns(SliceBehavior.JustInTime);
+        theRunner.ErrorHandlingOptions(Arg.Any<ShardExecutionMode>()).Returns(new ErrorHandlingOptions());
+    }
+
+    private async Task<RecordingLogger> executeRangeThatFailsDuringTeardownAsync(Exception failure)
+    {
+        var buildStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseBuild = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        async Task<IProjectionBatch> buildThenFailAsync()
+        {
+            buildStarted.TrySetResult();
+            await releaseBuild.Task;
+            throw failure;
+        }
+
+        theRunner.BuildBatchAsync(Arg.Any<EventRange>(), Arg.Any<ShardExecutionMode>(), Arg.Any<CancellationToken>())
+            .Returns(_ => buildThenFailAsync());
+
+        var execution = new GroupedProjectionExecution(new ShardName("Fake"), theRunner, theLogger);
+
+        var page = new EventPage(0);
+        page.CalculateCeiling(500, 10);
+        await execution.EnqueueAsync(page, theAgent);
+
+        await buildStarted.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+        // Teardown starts while the batch build is in flight -- the exact window jasperfx#507
+        // is about
+        await execution.HardStopAsync();
+        releaseBuild.SetResult();
+
+        // The failure surfaces asynchronously on the block's consumer; wait for the first log
+        // entry carrying it (buildBatchAsync logs at Error before rethrowing into the teardown
+        // clause), then give the same-continuation teardown clause a beat to write its own entry
+        var deadline = DateTimeOffset.UtcNow.AddSeconds(5);
+        while (DateTimeOffset.UtcNow < deadline)
+        {
+            lock (theLogger.Entries)
+            {
+                if (theLogger.Entries.Any(x => ReferenceEquals(x.Exception, failure)))
+                {
+                    break;
+                }
+            }
+
+            await Task.Delay(25);
+        }
+
+        await Task.Delay(250);
+
+        return theLogger;
+    }
+
+    [Fact]
+    public async Task a_real_database_failure_during_teardown_is_logged_not_discarded()
+    {
+        var schemaProblem = new FakeDbException("23514");
+
+        var logger = await executeRangeThatFailsDuringTeardownAsync(schemaProblem);
+
+        // A 23514 is a schema problem regardless of the CTS state (jasperfx#507). buildBatchAsync
+        // logs it at Error on the way out; the teardown clause records it again instead of
+        // discarding it into the void
+        logger.Entries.Any(x =>
+                x.Level == LogLevel.Information &&
+                ReferenceEquals(x.Exception, schemaProblem) &&
+                x.Message.Contains("does not look like a cancellation side effect"))
+            .ShouldBeTrue(
+                $"expected an Information entry for the discarded schema problem, got: {logger.Describe()}");
+
+        // But teardown must still not promote the failure to a critical shard failure
+        await theAgent.DidNotReceive().ReportCriticalFailureAsync(Arg.Any<Exception>());
+    }
+
+    [Fact]
+    public async Task a_genuine_cancellation_during_teardown_stays_silent()
+    {
+        var cancellation = new OperationCanceledException();
+
+        var logger = await executeRangeThatFailsDuringTeardownAsync(cancellation);
+
+        logger.Entries.Any(x => x.Message.Contains("does not look like a cancellation side effect"))
+            .ShouldBeFalse($"a pure cancellation must not be logged as a discarded failure, got: {logger.Describe()}");
+
+        await theAgent.DidNotReceive().ReportCriticalFailureAsync(Arg.Any<Exception>());
+    }
+}
+
+public class FakeDbException : DbException
+{
+    public FakeDbException(string? sqlState)
+    {
+        SqlState = sqlState;
+    }
+
+    public override string? SqlState { get; }
+}
+
+public class RecordingLogger : ILogger
+{
+    public record Entry(LogLevel Level, string Message, Exception? Exception);
+
+    public List<Entry> Entries { get; } = new();
+
+    public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+        Func<TState, Exception?, string> formatter)
+    {
+        lock (Entries)
+        {
+            Entries.Add(new Entry(logLevel, formatter(state, exception), exception));
+        }
+    }
+
+    public bool IsEnabled(LogLevel logLevel)
+    {
+        return true;
+    }
+
+    public IDisposable? BeginScope<TState>(TState state) where TState : notnull
+    {
+        return null;
+    }
+
+    public string Describe()
+    {
+        lock (Entries)
+        {
+            return string.Join("\n", Entries.Select(x => $"{x.Level}: {x.Message} ({x.Exception?.GetType().Name})"));
+        }
+    }
+}
diff --git a/src/JasperFx.Events/Daemon/CancellationExceptions.cs b/src/JasperFx.Events/Daemon/CancellationExceptions.cs
new file mode 100644
index 00000000..eb3872e8
--- /dev/null
+++ b/src/JasperFx.Events/Daemon/CancellationExceptions.cs
@@ -0,0 +1,49 @@
+using System.Data.Common;
+
+namespace JasperFx.Events.Daemon;
+
+/// <summary>
+///     Classifies whether an exception observed while a shard is being cancelled or torn down is a
+///     genuine side effect of that cancellation, or a real failure that deserves to be logged before
+///     the teardown discards it. A schema problem doesn't stop being a schema problem just because
+///     the shard's CancellationTokenSource fired first (jasperfx#507)
+/// </summary>
+public static class CancellationExceptions
+{
+    public static bool IsCancellationLike(Exception exception)
+    {
+        switch (exception)
+        {
+            case OperationCanceledException:
+                return true;
+
+            // A connection/session disposed underneath an in-flight operation during teardown
+            case ObjectDisposedException:
+                return true;
+
+            case AggregateException aggregate:
+                return aggregate.InnerExceptions.Count > 0 &&
+                       aggregate.InnerExceptions.All(IsCancellationLike);
+
+            // Provider-agnostic via DbException.SqlState (populated by Npgsql among others).
+            // Database exceptions like MartenCommandException wrap the provider exception, so the
+            // inner-exception walk below covers the wrapped case
+            case DbException db when isCancellationSqlState(db.SqlState):
+                return true;
+        }
+
+        return exception.InnerException != null && IsCancellationLike(exception.InnerException);
+    }
+
+    private static bool isCancellationSqlState(string? sqlState)
+    {
+        if (sqlState == null)
+        {
+            return false;
+        }
+
+        // 57014 = query_canceled (the server-side face of a cancelled command);
+        // class 08 = connection exceptions (connection_failure, connection_does_not_exist, ...)
+        return sqlState == "57014" || sqlState.StartsWith("08", StringComparison.Ordinal);
+    }
+}
diff --git a/src/JasperFx.Events/Daemon/GroupedProjectionExecution.cs b/src/JasperFx.Events/Daemon/GroupedProjectionExecution.cs
index 08360bdc..38fdf41a 100644
--- a/src/JasperFx.Events/Daemon/GroupedProjectionExecution.cs
+++ b/src/JasperFx.Events/Daemon/GroupedProjectionExecution.cs
@@ -20,11 +20,39 @@ public GroupedProjectionExecution(ShardName shardName, IGroupedProjectionRunner
         _logger = logger;
 
         var block = new Block<EventRange>(processRangeAsync);
+        block.OnError = onBlockFailure;
         _grouping = block.PushUpstream<EventRange>(groupEventRangeAsync);
+        _grouping.OnError = onBlockFailure;
 
         _runner = runner;
     }
 
+    // Last-resort sink for exceptions that escape processRangeAsync/groupEventRangeAsync (which
+    // have their own error handling) or that fault the block itself. Without this, such failures
+    // fell into Block<T>'s invisible default error handler and the shard died with zero log
+    // output (jasperfx#506)
+    private void onBlockFailure(EventRange? range, Exception ex)
+    {
+        _logger.LogError(ex, "Exception escaped the projection execution block for shard {Name}",
+            ShardName.Identity);
+
+        if (range?.Agent != null)
+        {
+            _ = Task.Run(async () =>
+            {
+                try
+                {
+                    await range.Agent.ReportCriticalFailureAsync(ex).ConfigureAwait(false);
+                }
+                catch (Exception reportingException)
+                {
+                    _logger.LogError(reportingException,
+                        "Failure while reporting a critical failure for shard {Name}", ShardName.Identity);
+                }
+            });
+        }
+    }
+
     public ShardName ShardName { get; }
 
     public object[]? Disposables { get; init; }
@@ -128,9 +156,18 @@ private async Task<EventRange> groupEventRangeAsync(EventRange range, Cancellati
 
             return range;
         }
-        catch when (_cancellation.IsCancellationRequested)
+        catch (Exception e) when (_cancellation.IsCancellationRequested)
         {
-            // Shard is being torn down — don't promote the cancellation to a critical failure.
+            // Shard is being torn down — don't promote a cancellation side effect to a critical
+            // failure. Anything that is NOT a genuine cancellation side effect is still logged
+            // before the teardown discards it (jasperfx#507)
+            if (!CancellationExceptions.IsCancellationLike(e))
+            {
+                _logger.LogInformation(e,
+                    "Discarding a failure while grouping events for {Name} because the shard is being stopped, but the exception does not look like a cancellation side effect",
+                    ShardName.Identity);
+            }
+
             return null!;
         }
         catch (Exception e)
@@ -188,16 +225,19 @@ private async Task processRangeAsync(EventRange range, CancellationToken _)
 
             range.Agent.Metrics.UpdateProcessed(range.Size);
         }
-        catch (OperationCanceledException) when (_cancellation.IsCancellationRequested)
+        catch (Exception e) when (_cancellation.IsCancellationRequested)
         {
             // Daemon-internal cancellation (StopAllAsync / HardStopAsync / DisposeAsync fired the
-            // shard's CTS). Don't surface as a critical shard failure — matches the guard already
-            // used in applyBatchOperationsToDatabaseAsync and SubscriptionExecutionBase.executeRange.
-        }
-        catch when (_cancellation.IsCancellationRequested)
-        {
-            // Wrapped/aggregated cancellation (e.g. Npgsql 57014 surfaced through a non-OCE) that
-            // is really a side effect of the shard being torn down. Same rationale as above.
+            // shard's CTS). A genuine cancellation side effect — OCE, wrapped/aggregated OCE, or a
+            // database exception whose SqlState is query-cancelled/connection-teardown — isn't a
+            // shard failure. Anything else (a schema problem is a schema problem regardless of the
+            // CTS state) is logged before the teardown discards it (jasperfx#507)
+            if (!CancellationExceptions.IsCancellationLike(e))
+            {
+                _logger.LogInformation(e,
+                    "Discarding a failure while processing events for {Name} because the shard is being stopped, but the exception does not look like a cancellation side effect",
+                    ShardName.Identity);
+            }
         }
         catch (Exception e)
         {
@@ -268,6 +308,13 @@ private async Task applyBatchOperationsToDatabaseAsync(EventRange range, IProjec
                     range);
                 throw;
             }
+
+            if (!CancellationExceptions.IsCancellationLike(e))
+            {
+                _logger.LogInformation(e,
+                    "Discarding a failure while executing an update batch for {Range} in shard '{Identity}' because the shard is being stopped, but the exception does not look like a cancellation side effect",
+                    range, ShardName.Identity);
+            }
         }
         finally
         {
diff --git a/src/JasperFx.Events/Daemon/ShardStateTracker.cs b/src/JasperFx.Events/Daemon/ShardStateTracker.cs
index 273f3946..1e5f015b 100644
--- a/src/JasperFx.Events/Daemon/ShardStateTracker.cs
+++ b/src/JasperFx.Events/Daemon/ShardStateTracker.cs
@@ -22,6 +22,8 @@ public ShardStateTracker(ILogger logger)
     {
         _logger = logger;
         _block = new Block<ShardState>(publish);
+        _block.OnError = (state, ex) =>
+            _logger.LogError(ex, "Failure while publishing shard state {State}", state);
 
         _subscription = Subscribe(this);
     }
diff --git a/src/JasperFx.Events/Daemon/SubscriptionAgent.cs b/src/JasperFx.Events/Daemon/SubscriptionAgent.cs
index 537dd8de..48b84a04 100644
--- a/src/JasperFx.Events/Daemon/SubscriptionAgent.cs
+++ b/src/JasperFx.Events/Daemon/SubscriptionAgent.cs
@@ -42,6 +42,30 @@ public SubscriptionAgent(ShardName name, AsyncOptions options, TimeProvider time
         _commandBlock = new Block<Command>(Apply);
 
         ProjectionShardIdentity = name.Identity;
+
+        // Last-resort sink for exceptions escaping the command loop. Without this, a failed
+        // command fell into Block<T>'s invisible default error handler and the agent silently
+        // stopped making progress (jasperfx#506). ReportCriticalFailureAsync does not post back
+        // onto the command block, so there is no recursion here
+        _commandBlock.OnError = (command, ex) =>
+        {
+            _logger.LogError(ex, "Error processing daemon command {Command} for shard {Identity}",
+                command, ProjectionShardIdentity);
+
+            _ = Task.Run(async () =>
+            {
+                try
+                {
+                    await ReportCriticalFailureAsync(ex).ConfigureAwait(false);
+                }
+                catch (Exception reportingException)
+                {
+                    _logger.LogError(reportingException,
+                        "Failure while reporting a critical failure for shard {Identity}",
+                        ProjectionShardIdentity);
+                }
+            });
+        };
     }
 
     // Exposed so tests can assert how the daemon composed this agent's loader (jasperfx#494)
diff --git a/src/JasperFx.Events/Daemon/SubscriptionExecutionBase.cs b/src/JasperFx.Events/Daemon/SubscriptionExecutionBase.cs
index ebbf12da..59104742 100644
--- a/src/JasperFx.Events/Daemon/SubscriptionExecutionBase.cs
+++ b/src/JasperFx.Events/Daemon/SubscriptionExecutionBase.cs
@@ -58,12 +58,37 @@ public SubscriptionExecutionBase(IEventDatabase database, ShardName name, ILogge
         _logger = logger;
 
         _executionBlock = new Block<EventRange>(executeRange);
+        _executionBlock.OnError = onBlockFailure;
 
-        // TODO -- revisit this. 
+        // TODO -- revisit this.
         ShardIdentity = $"{name.Identity}@{database.Identifier}";
         ShardName = name;
     }
 
+    // Last-resort sink for exceptions that escape executeRange (which has its own error handling)
+    // or that fault the block itself. Without this, such failures fell into Block<T>'s invisible
+    // default error handler and the subscription died with zero log output (jasperfx#506)
+    private void onBlockFailure(EventRange? range, Exception ex)
+    {
+        _logger.LogError(ex, "Exception escaped the subscription execution block for {Name}", ShardIdentity);
+
+        if (range?.Agent != null)
+        {
+            _ = Task.Run(async () =>
+            {
+                try
+                {
+                    await range.Agent.ReportCriticalFailureAsync(ex).ConfigureAwait(false);
+                }
+                catch (Exception reportingException)
+                {
+                    _logger.LogError(reportingException,
+                        "Failure while reporting a critical failure for {Name}", ShardIdentity);
+                }
+            });
+        }
+    }
+
     public ILogger? Logger { get; set; } = NullLogger.Instance;
 
     /// <summary>
@@ -168,6 +193,17 @@ private async Task executeRange(EventRange range, CancellationToken _)
         catch (OperationCanceledException)
         {
         }
+        catch (Exception e) when (_cancellation.IsCancellationRequested)
+        {
+            // Subscription is being torn down. A genuine cancellation side effect isn't a failure,
+            // but anything else is at least logged before the teardown discards it (jasperfx#507)
+            if (!CancellationExceptions.IsCancellationLike(e))
+            {
+                _logger.LogInformation(e,
+                    "Discarding a failure while processing subscription {Name} because the subscription is being stopped, but the exception does not look like a cancellation side effect",
+                    ShardIdentity);
+            }
+        }
         catch (Exception e)
         {
             activity?.AddException(e);
diff --git a/src/JasperFx.Events/Projections/ProjectionExecution.cs b/src/JasperFx.Events/Projections/ProjectionExecution.cs
index 23c77a73..cf74579e 100644
--- a/src/JasperFx.Events/Projections/ProjectionExecution.cs
+++ b/src/JasperFx.Events/Projections/ProjectionExecution.cs
@@ -31,6 +31,32 @@ public ProjectionExecution(ShardName shardName, AsyncOptions options,
         _logger = logger;
 
         _building = new Block<EventRange>(processRangeAsync);
+        _building.OnError = onBlockFailure;
+    }
+
+    // Last-resort sink for exceptions that escape processRangeAsync (which has its own error
+    // handling) or that fault the block itself. Without this, such failures fell into Block<T>'s
+    // invisible default error handler and the shard died with zero log output (jasperfx#506)
+    private void onBlockFailure(EventRange? range, Exception ex)
+    {
+        _logger.LogError(ex, "Exception escaped the projection execution block for shard {Name}",
+            _shardName.Identity);
+
+        if (range?.Agent != null)
+        {
+            _ = Task.Run(async () =>
+            {
+                try
+                {
+                    await range.Agent.ReportCriticalFailureAsync(ex).ConfigureAwait(false);
+                }
+                catch (Exception reportingException)
+                {
+                    _logger.LogError(reportingException,
+                        "Failure while reporting a critical failure for shard {Name}", _shardName.Identity);
+                }
+            });
+        }
     }
 
     public ShardName ShardName => _shardName;
@@ -119,6 +145,17 @@ private async Task processRangeAsync(EventRange range, CancellationToken _)
 
             range.Agent.Metrics.UpdateProcessed(range.Size);
         }
+        catch (Exception e) when (_cancellation.IsCancellationRequested)
+        {
+            // Shard is being torn down. A genuine cancellation side effect isn't a failure, but
+            // anything else is at least logged before the teardown discards it (jasperfx#507)
+            if (!CancellationExceptions.IsCancellationLike(e))
+            {
+                _logger.LogInformation(e,
+                    "Discarding a failure while processing events for {Name} because the shard is being stopped, but the exception does not look like a cancellation side effect",
+                    _shardName.Identity);
+            }
+        }
         catch (Exception e)
         {
             activity?.AddException(e);
@@ -197,6 +234,13 @@ private async Task applyBatchOperationsToDatabaseAsync(EventRange range, IProjec
                     range);
                 throw;
             }
+
+            if (!CancellationExceptions.IsCancellationLike(e))
+            {
+                _logger.LogInformation(e,
+                    "Discarding a failure while executing an update batch for {Range} in shard '{Identity}' because the shard is being stopped, but the exception does not look like a cancellation side effect",
+                    range, _shardName.Identity);
+            }
         }
         finally
         {
diff --git a/src/JasperFx/Blocks/BatchingChannel.cs b/src/JasperFx/Blocks/BatchingChannel.cs
index 972fdfd8..1e36b5d4 100644
--- a/src/JasperFx/Blocks/BatchingChannel.cs
+++ b/src/JasperFx/Blocks/BatchingChannel.cs
@@ -39,6 +39,12 @@ public BatchingChannel(TimeSpan timeOut, IBlock<T[]> downstream, int batchSize =
 
     public override uint Count => (uint)_current.Count + _inner.Count;
 
+    public override Action<T, Exception> OnError
+    {
+        get => _inner.OnError;
+        set => _inner.OnError = value;
+    }
+
     public void TriggerBatch()
     {
         lock (_syncLock)
diff --git a/src/JasperFx/Blocks/Block.cs b/src/JasperFx/Blocks/Block.cs
index 0b7c4a10..18a5923e 100644
--- a/src/JasperFx/Blocks/Block.cs
+++ b/src/JasperFx/Blocks/Block.cs
@@ -25,6 +25,7 @@ public class Block<T> : BlockBase<T>
     private readonly Task[] _tasks;
     private bool _latched;
     private uint _count = 0;
+    private Exception? _failure;
 
     public Block(Func<T, CancellationToken, Task> action) : this(1, action)
     {
@@ -82,18 +83,30 @@ public Block(int parallelCount, int boundedCapacity, Func<T, CancellationToken,
         }
     }
 
+    // Deliberately NOT Debug.WriteLine: that call is compiled away in release builds, which made
+    // any exception escaping a block action completely invisible in production (jasperfx#506).
+    // This is a last-resort sink -- anything hosting a Block should assign OnError to real logging.
     private Action<T, Exception> _onError = (item, ex) =>
     {
-        Debug.WriteLine("Error processing item " + item);
-        Debug.WriteLine(ex.ToString());
+        Console.Error.WriteLine($"JasperFx Block<{typeof(T).Name}>: error processing item {item}");
+        Console.Error.WriteLine(ex.ToString());
+        Trace.WriteLine(ex.ToString());
     };
 
-    public Action<T, Exception> OnError
+    public override Action<T, Exception> OnError
     {
         get => _onError;
         set => _onError = value ?? throw new ArgumentNullException(nameof(OnError));
     }
 
+    /// <summary>
+    /// The terminal exception that faulted this block, if any. A faulted block has permanently
+    /// stopped processing and throws from <see cref="Post"/>/<see cref="PostAsync"/> so that a
+    /// dead consumer is observable by its producers instead of silently accumulating items
+    /// (jasperfx#506)
+    /// </summary>
+    public Exception? Failure => Volatile.Read(ref _failure);
+
     public override async Task WaitForCompletionAsync()
     {
         Complete();
@@ -102,6 +115,11 @@ public override async Task WaitForCompletionAsync()
 
         await Task.WhenAll(_tasks);
 
+        // A faulted block must not try to drain -- its action already proved terminally broken.
+        // Producers learn about the fault from Post/PostAsync; teardown paths calling this method
+        // should still complete quietly.
+        if (_failure != null) return;
+
         while (!Cancellation.IsCancellationRequested && _count > 0)
         {
             try
@@ -143,34 +161,96 @@ public override async Task WaitForCompletionAsync()
 
     private async Task processAsync()
     {
-        while (!_cancellation.IsCancellationRequested)
+        try
         {
-            var isData = await _channel.Reader.WaitToReadAsync(_cancellation.Token);
-            if (!isData) return;
-
-            if (_channel.Reader.TryRead(out var item))
+            while (!_cancellation.IsCancellationRequested)
             {
-                try
-                {
-                    await _action(item, _cancellation.Token);
-                }
-                catch (Exception e)
-                {
-                    _onError(item, e);
-                }
-                finally
+                var isData = await _channel.Reader.WaitToReadAsync(_cancellation.Token);
+                if (!isData) return;
+
+                if (_channel.Reader.TryRead(out var item))
                 {
-                    Interlocked.Decrement(ref _count);
+                    try
+                    {
+                        await _action(item, _cancellation.Token);
+                    }
+                    catch (Exception e)
+                    {
+                        // The error callback itself must never be able to kill the consumer loop.
+                        // If it throws, error handling is terminally broken and the only honest
+                        // move is to fault the whole block (jasperfx#506)
+                        try
+                        {
+                            _onError(item, e);
+                        }
+                        catch (Exception handlerException)
+                        {
+                            Interlocked.Decrement(ref _count);
+                            fault(new AggregateException(e, handlerException));
+                            return;
+                        }
+                    }
+                    finally
+                    {
+                        Interlocked.Decrement(ref _count);
+                    }
                 }
+
+                if (_latched) return;
             }
-            
-            
-            if (_latched) return;
+        }
+        catch (OperationCanceledException)
+        {
+            // Normal block shutdown
+        }
+        catch (ObjectDisposedException)
+        {
+            // The block itself was disposed out from under the consumer loop
+        }
+        catch (Exception e)
+        {
+            // The consumer loop died on something other than shutdown. The old behavior was to
+            // exit permanently while Post/PostAsync kept accepting items that would never be
+            // processed (jasperfx#506) -- instead, fault the block so producers find out
+            fault(e);
+        }
+    }
+
+    private void fault(Exception failure)
+    {
+        if (Interlocked.CompareExchange(ref _failure, failure, null) != null) return;
+
+        _latched = true;
+
+        // Completing the channel with the failure releases any producer currently block-waiting
+        // for capacity in Post/PostAsync with a ChannelClosedException instead of hanging forever
+        // against a consumer that will never drain the channel
+        _channel.Writer.TryComplete(failure);
+
+        try
+        {
+            _onError(default!, failure);
+        }
+        catch (Exception)
+        {
+            // The error callback already proved unreliable; there is nothing else to do with this
+        }
+    }
+
+    private void assertNotFaulted()
+    {
+        if (_failure != null)
+        {
+            throw new InvalidOperationException(
+                $"This Block<{typeof(T).Name}> has faulted and can no longer accept items. See the inner exception for the terminal failure.",
+                _failure);
         }
     }
 
     public override void Post(T item)
     {
+        assertNotFaulted();
+
         if (_latched) return;
 
         Interlocked.Increment(ref _count);
@@ -191,21 +271,28 @@ public override void Post(T item)
         catch (Exception e)
         {
             Interlocked.Decrement(ref _count);
+
+            // A producer released from the blocking write by a fault must observe the fault, not a
+            // swallowed error callback (jasperfx#506)
+            assertNotFaulted();
+
             try
             {
                 _onError(item, e);
             }
             catch (Exception)
             {
-                Debug.WriteLine($"Was not able to write {item} to the queue synchronously!");
+                Trace.WriteLine($"Was not able to write {item} to the queue synchronously!");
             }
         }
     }
 
     public override ValueTask PostAsync(T item)
     {
+        assertNotFaulted();
+
         if (_latched) return ValueTask.CompletedTask;
-        
+
         Interlocked.Increment(ref _count);
         return _channel.Writer.WriteAsync(item, _cancellation.Token);
     }
diff --git a/src/JasperFx/Blocks/BlockBase.cs b/src/JasperFx/Blocks/BlockBase.cs
index 8161ff9b..0dcedf68 100644
--- a/src/JasperFx/Blocks/BlockBase.cs
+++ b/src/JasperFx/Blocks/BlockBase.cs
@@ -8,6 +8,14 @@ public abstract class BlockBase<T> : IBlock<T>
     public abstract ValueTask PostAsync(T item);
     public abstract void Post(T item);
 
+    /// <summary>
+    /// Error callback for exceptions escaping this block's processing. Blocks that execute an
+    /// action directly (like <see cref="Block{T}"/>) invoke this; composite blocks delegate it to
+    /// their processing block. The default is a no-op holder for subclasses that never execute
+    /// user actions themselves
+    /// </summary>
+    public virtual Action<T, Exception> OnError { get; set; } = (_, _) => { };
+
     public IBlock<TBefore> PushUpstream<TBefore>(Func<TBefore, CancellationToken, Task<T>> transformation)
     {
         var top = new Block<TBefore>(async (item, token) =>
diff --git a/src/JasperFx/Blocks/BlockSet.cs b/src/JasperFx/Blocks/BlockSet.cs
index 05aeb867..7a2244d0 100644
--- a/src/JasperFx/Blocks/BlockSet.cs
+++ b/src/JasperFx/Blocks/BlockSet.cs
@@ -24,6 +24,16 @@ public uint Count
         }
     }
 
+    /// <summary>
+    /// Delegates to the top block of the set -- the one whose processing action runs first for
+    /// items posted to this set
+    /// </summary>
+    public Action<T, Exception> OnError
+    {
+        get => _top.OnError;
+        set => _top.OnError = value;
+    }
+
     public IBlock<TBefore> PushUpstream<TBefore>(Func<TBefore, CancellationToken, Task<T>> transformation)
     {
         var top = new Block<TBefore>(async (item, token) =>
diff --git a/src/JasperFx/Blocks/IBlock.cs b/src/JasperFx/Blocks/IBlock.cs
index 7a2ff356..7b5f33b1 100644
--- a/src/JasperFx/Blocks/IBlock.cs
+++ b/src/JasperFx/Blocks/IBlock.cs
@@ -18,6 +18,12 @@ public interface IBlock : IAsyncDisposable
 /// <typeparam name="T"></typeparam>
 public interface IBlock<T> : IBlock
 {
+    /// <summary>
+    /// Error callback for exceptions escaping this block's processing action. Hosts should assign
+    /// this to real logging -- the default sink is a last resort (jasperfx#506)
+    /// </summary>
+    Action<T, Exception> OnError { get; set; }
+
     /// <summary>
     /// Add a new item to this item to be processed in the background
     /// </summary>