diff --git a/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs
new file mode 100644
index 00000000..ed496fc0
--- /dev/null
+++ b/src/Wolfgang.Etl.Abstractions/EtlPipeline/EtlPipelineSinkExtensions.cs
@@ -0,0 +1,98 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+
+namespace Wolfgang.Etl.Abstractions;
+
+///
+/// Extensions for terminal pipelines.
+///
+public static class EtlPipelineSinkExtensions
+{
+ ///
+ /// Wraps a sink so that the given resources — opened by the code that built the pipeline (for
+ /// example a file stream a path-based loader factory created) — are disposed after the run
+ /// completes, whether it succeeds or throws.
+ ///
+ /// The sink to wrap.
+ ///
+ /// The resources to dispose after the run. Each is disposed via
+ /// when supported, otherwise ; anything else (or )
+ /// is skipped. Resources are disposed in reverse of the order supplied (LIFO).
+ ///
+ ///
+ /// A sink that disposes after running, or the original
+ /// unchanged when no resources are supplied.
+ ///
+ /// is .
+ public static IEtlPipelineSink DisposingOwned
+ (
+ this IEtlPipelineSink sink,
+ params object[] ownedResources
+ )
+ {
+ if (sink is null)
+ {
+ throw new ArgumentNullException(nameof(sink));
+ }
+
+ if (ownedResources is null || ownedResources.Length == 0)
+ {
+ return sink;
+ }
+
+ return new DisposingSink(sink, ownedResources);
+ }
+
+
+ private sealed class DisposingSink : IEtlPipelineSink
+ {
+ private readonly IEtlPipelineSink _inner;
+ private readonly object[] _ownedResources;
+
+
+ internal DisposingSink(IEtlPipelineSink inner, object[] ownedResources)
+ {
+ _inner = inner;
+ _ownedResources = ownedResources;
+ }
+
+
+ public async Task RunAsync
+ (
+ IProgress? progress = null,
+ CancellationToken token = default
+ )
+ {
+ try
+ {
+ await _inner.RunAsync(progress, token).ConfigureAwait(false);
+ }
+ finally
+ {
+ await DisposeOwnedAsync().ConfigureAwait(false);
+ }
+ }
+
+
+ // Reverse (LIFO) order, matching nested using / DI-scope disposal. IAsyncDisposable is
+ // preferred where the concrete resource implements it (e.g. FileStream on net5.0+); on the
+ // older frameworks the same FileStream matches only IDisposable, so it disposes synchronously.
+ private async Task DisposeOwnedAsync()
+ {
+ for (var i = _ownedResources.Length - 1; i >= 0; i--)
+ {
+ switch (_ownedResources[i])
+ {
+ case IAsyncDisposable asyncDisposable:
+ await asyncDisposable.DisposeAsync().ConfigureAwait(false);
+ break;
+ case IDisposable disposable:
+ disposable.Dispose();
+ break;
+ }
+ }
+ }
+ }
+}
diff --git a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt
index 71f7efaa..88c5115f 100644
--- a/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt
+++ b/src/Wolfgang.Etl.Abstractions/PublicAPI.Shipped.txt
@@ -170,3 +170,5 @@ virtual Wolfgang.Etl.Abstractions.TransformerBase.TransformAsync(System.Collections.Generic.IAsyncEnumerable! items, System.IProgress! progress) -> System.Collections.Generic.IAsyncEnumerable!
virtual Wolfgang.Etl.Abstractions.TransformerBase.TransformAsync(System.Collections.Generic.IAsyncEnumerable! items, System.IProgress! progress, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable!
virtual Wolfgang.Etl.Abstractions.TransformerBase.TransformAsync(System.Collections.Generic.IAsyncEnumerable! items, System.Threading.CancellationToken token) -> System.Collections.Generic.IAsyncEnumerable!
+Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions
+static Wolfgang.Etl.Abstractions.EtlPipelineSinkExtensions.DisposingOwned(this Wolfgang.Etl.Abstractions.IEtlPipelineSink! sink, params object![]! ownedResources) -> Wolfgang.Etl.Abstractions.IEtlPipelineSink!
diff --git a/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs
new file mode 100644
index 00000000..88864d08
--- /dev/null
+++ b/tests/Wolfgang.Etl.Abstractions.Tests.Unit/EtlPipelineTests/EtlPipelineSinkExtensionsTests.cs
@@ -0,0 +1,121 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace Wolfgang.Etl.Abstractions.Tests.Unit.EtlPipelineTests;
+
+public sealed class EtlPipelineSinkExtensionsTests
+{
+ private sealed class FakeSink : IEtlPipelineSink
+ {
+ private readonly Func? _onRun;
+
+ public FakeSink(Func? onRun = null) => _onRun = onRun;
+
+ public bool Ran { get; private set; }
+
+ public async Task RunAsync(IProgress? progress = null, CancellationToken token = default)
+ {
+ Ran = true;
+ if (_onRun is not null)
+ {
+ await _onRun().ConfigureAwait(false);
+ }
+ }
+ }
+
+
+ private sealed class TrackingDisposable : IDisposable
+ {
+ private readonly List _log;
+ private readonly string _name;
+
+ public TrackingDisposable(List log, string name)
+ {
+ _log = log;
+ _name = name;
+ }
+
+ public void Dispose() => _log.Add(_name);
+ }
+
+
+ private sealed class TrackingAsyncDisposable : IAsyncDisposable
+ {
+ private readonly List _log;
+ private readonly string _name;
+
+ public TrackingAsyncDisposable(List log, string name)
+ {
+ _log = log;
+ _name = name;
+ }
+
+ public ValueTask DisposeAsync()
+ {
+ _log.Add(_name);
+ return default;
+ }
+ }
+
+
+ [Fact]
+ public async Task DisposingOwned_disposes_resources_after_a_successful_run()
+ {
+ var log = new List();
+ var sink = new FakeSink();
+
+ await sink.DisposingOwned(new TrackingDisposable(log, "a")).RunAsync();
+
+ Assert.True(sink.Ran);
+ Assert.Equal(new[] { "a" }, log);
+ }
+
+
+ [Fact]
+ public async Task DisposingOwned_disposes_in_reverse_order()
+ {
+ var log = new List();
+ var sink = new FakeSink();
+
+ await sink
+ .DisposingOwned(
+ new TrackingDisposable(log, "first"),
+ new TrackingAsyncDisposable(log, "second"))
+ .RunAsync();
+
+ Assert.Equal(new[] { "second", "first" }, log);
+ }
+
+
+ [Fact]
+ public async Task DisposingOwned_disposes_even_when_the_run_throws()
+ {
+ var log = new List();
+ var sink = new FakeSink(() => throw new InvalidOperationException("boom"));
+
+ await Assert.ThrowsAsync(
+ () => sink.DisposingOwned(new TrackingDisposable(log, "a")).RunAsync());
+
+ Assert.Equal(new[] { "a" }, log);
+ }
+
+
+ [Fact]
+ public void DisposingOwned_with_no_resources_returns_the_same_sink()
+ {
+ var sink = new FakeSink();
+
+ Assert.Same(sink, sink.DisposingOwned());
+ }
+
+
+ [Fact]
+ public void DisposingOwned_null_sink_throws()
+ {
+ Assert.Throws(
+ () => ((IEtlPipelineSink)null!).DisposingOwned(new object()));
+ }
+}