Implements IJobWorker.IAsyncDisposable #771

xlegalles · 2025-02-10T14:04:04Z

#769
There is a slight behavior difference as the isRunning flag will only be set when the worker is effectively closed.

ChrisKujawa

Thanks 🚀

ChrisKujawa · 2025-02-10T16:13:12Z

Client/Impl/Worker/JobWorker.cs

+    [Obsolete("Use DisposeAsync instead.", false)]
    public void Dispose()
+    {
+        _ = DisposeAsync();


❓️ should we wait here?

The current implementation does not wait. This is what is problematic. But waiting an async method from a sync one is really not a good practice

After reading the docs:

https://learn.microsoft.com/en-us/dotnet/standard/garbage-collection/implementing-dispose

https://learn.microsoft.com/en-us/dotnet/api/system.iasyncdisposable?view=net-9.0

I thought we might want to write the methods a bit differently and make sure we close/dispose all resources more directly. Especially after the reformat we see that there several objects, we are not keeping the reference to dispose them correctly. Speaking of the _. https://github.com/camunda-community-hub/zeebe-client-csharp/blob/main/Client/Impl/Worker/JobWorker.cs#L107-L111

I guess we could simply call dispose on them, like on the Task thats starts the polling, etc.

But I guess we still need the cancelation of the source - and disposing of it

xlegalles · 2025-02-11T11:45:53Z

Client/Impl/Worker/JobWorker.cs

-        _ = transformer.LinkTo(output);
+        var linkInputTransformer = input.LinkTo(transformer);
+        var linkTransformerOutput = transformer.LinkTo(output);



Track polling Task in order to await for its completion. Note that I do not want to cancel the continuation. Finally, dispose TPL block objects.

xlegalles · 2025-02-11T11:48:05Z

Client/Impl/Worker/JobWorker.cs

            }
        }
    }



Update the signature so that we can pass the cancellation token to SendAsync

xlegalles · 2025-02-11T11:49:39Z

Client/Impl/Worker/JobWorker.cs


    /// <inheritdoc />
    public async ValueTask DisposeAsync()
    {


Cancel the source token which will stop the pollingTask. Note that I've changed the log because it's not an error and very annoying to see it in reports.

ChrisKujawa · 2025-02-19T07:48:42Z

Client/Impl/Worker/JobWorker.cs

+
+            linkInputTransformer.Dispose();
+            linkTransformerOutput.Dispose();


❓ Should we maybe dispose the objects in our dispose? But I guess this will not work in the error case.

Client.UnitTests/JobWorkerTest.cs

Copilot

Pull request overview

This PR implements the IAsyncDisposable interface for IJobWorker, enabling proper asynchronous cleanup of job workers. The synchronous Dispose() method is marked as obsolete to encourage migration to the async disposal pattern. The implementation now properly awaits the polling task completion during disposal and cleans up dataflow links.

Key changes:

Added IAsyncDisposable to the IJobWorker interface
Implemented DisposeAsync() that cancels polling and waits for task completion
Enhanced cleanup logic to dispose dataflow pipeline links in a continuation

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 8 comments.

File	Description
`Client/Api/Worker/IJobWorker.cs`	Extended interface to implement `IAsyncDisposable` alongside existing `IDisposable`
`Client/Impl/Worker/JobWorker.cs`	Implemented async disposal with proper polling task cancellation and awaiting; enabled nullable reference types; improved resource cleanup by disposing dataflow links; marked synchronous `Dispose()` as obsolete; removed discard operators for better code clarity
`Client.UnitTests/JobWorkerTest.cs`	Added test case for async disposal of running workers; removed unnecessary discard operators throughout existing tests; improved lambda parameter naming for unused parameters; updated log message to use structured logging format

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-07T17:38:40Z

Client/Impl/Worker/JobWorker.cs

+            source.Cancel();
+            await pollingTask;
+        }
+
+        logger?.LogInformation("JobWorker is now disposed");
+        source.Dispose();
        isRunning = false;


After source.Dispose() is called on line 81, the CancellationTokenSource is disposed but isRunning is still true momentarily. If another thread calls Open() during this window, it would attempt to access source.Token on line 104, which would throw an ObjectDisposedException.

The order of operations should be:

Set isRunning = false first (to prevent new Open() calls)

Cancel the token

Await the polling task

Dispose the source

Alternatively, add a check in Open() to prevent re-opening after disposal.

Suggested change

source.Cancel();

await pollingTask;

}

logger?.LogInformation("JobWorker is now disposed");

source.Dispose();

isRunning = false;

isRunning = false;

source.Cancel();

await pollingTask;

}

logger?.LogInformation("JobWorker is now disposed");

source.Dispose();

Copilot · 2025-12-07T17:38:40Z

Client/Impl/Worker/JobWorker.cs

-        // delay disposing, since poll and handler take some time to close
-        _ = Task.Delay(TimeSpan.FromMilliseconds(pollInterval.TotalMilliseconds * 2))
-        .ContinueWith(t =>
+        _ = DisposeAsync();


The synchronous Dispose() method discards the ValueTask returned by DisposeAsync() without awaiting it. This means the method returns before disposal is complete, potentially leaving resources in an inconsistent state. According to C# best practices, synchronous disposal should block until async disposal completes. Consider implementing this as:

public void Dispose() { DisposeAsync().AsTask().GetAwaiter().GetResult(); }

This ensures the synchronous Dispose method blocks until disposal is complete, maintaining proper resource cleanup semantics.

Suggested change

_ = DisposeAsync();

DisposeAsync().AsTask().GetAwaiter().GetResult();

Copilot · 2025-12-07T17:38:41Z

Client/Impl/Worker/JobWorker.cs

+    public async ValueTask DisposeAsync()
+    {
+        if (pollingTask != null)
        {
-            logger?.LogError("Dispose source");
-            source.Dispose();
-        });
+            source.Cancel();
+            await pollingTask;
+        }
+
+        logger?.LogInformation("JobWorker is now disposed");
+        source.Dispose();
        isRunning = false;
    }


DisposeAsync() is not thread-safe and can be called multiple times concurrently, leading to race conditions. The method should:

Check if already disposed and return early

Ensure thread-safe disposal (e.g., using Interlocked operations or a lock)

Only dispose source once

Consider adding a disposed flag and using Interlocked.CompareExchange to ensure single execution:

private int disposed = 0; public async ValueTask DisposeAsync() { if (Interlocked.CompareExchange(ref disposed, 1, 0) == 1) { return; // Already disposed } if (pollingTask != null) { source.Cancel(); await pollingTask; } logger?.LogInformation("JobWorker is now disposed"); source.Dispose(); isRunning = false; }

Copilot · 2025-12-07T17:38:41Z

Client/Impl/Worker/JobWorker.cs

+
+        logger?.LogInformation("JobWorker is now disposed");
+        source.Dispose();
        isRunning = false;


The isRunning flag is set to false after source.Dispose(), but the IsOpen() and IsClosed() methods read this volatile field without synchronization with the disposal process. This creates a race condition where:

A thread could check IsOpen() and see true

DisposeAsync completes on another thread

The first thread tries to use the worker assuming it's still open

The isRunning flag should be set to false before cancellation and disposal begin, not after. Consider moving line 82 to the beginning of the method (after the disposed check if one is added).

Copilot · 2025-12-07T17:38:41Z

Client/Impl/Worker/JobWorker.cs

                {
                    if (task.IsFaulted)
                    {
                        logger?.LogWarning(task.Exception, "Problem on failing job occured.");


Typo in log message: "occured" should be "occurred".

Suggested change

logger?.LogWarning(task.Exception, "Problem on failing job occured.");

logger?.LogWarning(task.Exception, "Problem on failing job occurred.");

Copilot · 2025-12-07T17:38:41Z

Client/Impl/Worker/JobWorker.cs

+                logger?.LogError(t.Exception, "Job polling failed");
+            }
+            else if (t.IsCanceled)
+            {
+                logger?.LogInformation("Job polling Cancelled");


[nitpick] Inconsistent punctuation in log messages. Line 125 uses "Job polling failed" without a period, but line 129 uses "Job polling Cancelled" (also without a period). However, this is inconsistent with the pattern in line 271 which includes a period. Consider standardizing log message punctuation across the codebase - either always include or always omit terminal periods.

Suggested change

logger?.LogError(t.Exception, "Job polling failed");

}

else if (t.IsCanceled)

{

logger?.LogInformation("Job polling Cancelled");

logger?.LogError(t.Exception, "Job polling failed.");

}

else if (t.IsCanceled)

{

logger?.LogInformation("Job polling Cancelled.");

Copilot · 2025-12-07T17:38:41Z

Client/Impl/Worker/JobWorker.cs

+    public async ValueTask DisposeAsync()
+    {
+        if (pollingTask != null)
        {
-            logger?.LogError("Dispose source");
-            source.Dispose();
-        });
+            source.Cancel();
+            await pollingTask;
+        }
+
+        logger?.LogInformation("JobWorker is now disposed");
+        source.Dispose();
        isRunning = false;
    }


The new DisposeAsync() implementation lacks test coverage for important edge cases:

Multiple concurrent calls to DisposeAsync()

Calling DisposeAsync() when the worker was never opened (pollingTask is null)

Calling both Dispose() and DisposeAsync() concurrently

The existing test CanDisposeRunningWorker() only covers the happy path. Consider adding tests for these scenarios to ensure thread-safety and idempotency.

Copilot · 2025-12-07T17:38:42Z

Client.UnitTests/JobWorkerTest.cs

+        var signal = new EventWaitHandle(false, EventResetMode.AutoReset);
+
+        // when
+        var jobWorker = ZeebeClient.NewWorker()
+            .JobType("foo")
+            .Handler((_, _) =>
+            {
+                // trigger worker disposal with second job
+                signal.Set();
+            })
+            .AutoCompletion()
+            .MaxJobsActive(3)
+            .Name("jobWorker")
+            .Timeout(TimeSpan.FromSeconds(1))
+            .PollInterval(TimeSpan.FromSeconds(3)) // long
+            .PollingTimeout(TimeSpan.FromMilliseconds(100))
+            .Open();
+        Assert.True(jobWorker.IsOpen());
+
+        signal.WaitOne();
+        // disposal must be quick even though the polling interval is long
+        await jobWorker.DisposeAsync().ConfigureAwait(false);
+        Assert.True(jobWorker.IsClosed());


Disposable 'EventWaitHandle' is created but not disposed.

Suggested change

var signal = new EventWaitHandle(false, EventResetMode.AutoReset);

// when

var jobWorker = ZeebeClient.NewWorker()

.JobType("foo")

.Handler((_, _) =>

{

// trigger worker disposal with second job

signal.Set();

})

.AutoCompletion()

.MaxJobsActive(3)

.Name("jobWorker")

.Timeout(TimeSpan.FromSeconds(1))

.PollInterval(TimeSpan.FromSeconds(3)) // long

.PollingTimeout(TimeSpan.FromMilliseconds(100))

.Open();

Assert.True(jobWorker.IsOpen());

signal.WaitOne();

// disposal must be quick even though the polling interval is long

await jobWorker.DisposeAsync().ConfigureAwait(false);

Assert.True(jobWorker.IsClosed());

using (var signal = new EventWaitHandle(false, EventResetMode.AutoReset))

{

// when

var jobWorker = ZeebeClient.NewWorker()

.JobType("foo")

.Handler((_, _) =>

{

// trigger worker disposal with second job

signal.Set();

})

.AutoCompletion()

.MaxJobsActive(3)

.Name("jobWorker")

.Timeout(TimeSpan.FromSeconds(1))

.PollInterval(TimeSpan.FromSeconds(3)) // long

.PollingTimeout(TimeSpan.FromMilliseconds(100))

.Open();

Assert.True(jobWorker.IsOpen());

signal.WaitOne();

// disposal must be quick even though the polling interval is long

await jobWorker.DisposeAsync().ConfigureAwait(false);

Assert.True(jobWorker.IsClosed());

}

ChrisKujawa · 2025-12-07T18:10:19Z

@nloding lets look at this next together

feat: implements IJobWorker.IAsyncDisposable

b7ec802

xlegalles requested a review from ChrisKujawa as a code owner February 10, 2025 14:04

ChrisKujawa approved these changes Feb 10, 2025

View reviewed changes

feat: PR comments. Improve disposal + UT

cc37a09

xlegalles commented Feb 11, 2025

View reviewed changes

Client/Impl/Worker/JobWorker.cs

}

}

}

Copy link

Contributor Author

xlegalles Feb 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Update the signature so that we can pass the cancellation token to SendAsync

xlegalles commented Feb 11, 2025

View reviewed changes

ChrisKujawa reviewed Feb 19, 2025

View reviewed changes

Client.UnitTests/JobWorkerTest.cs Outdated Show resolved Hide resolved

fix PR comments

f2b6276

ChrisKujawa requested a review from Copilot December 7, 2025 17:34

Copilot started reviewing on behalf of ChrisKujawa December 7, 2025 17:35 View session

Copilot AI reviewed Dec 7, 2025

View reviewed changes

ChrisKujawa requested a review from nloding December 7, 2025 18:10


		linkInputTransformer.Dispose();
		linkTransformerOutput.Dispose();

	_ = DisposeAsync();
	DisposeAsync().AsTask().GetAwaiter().GetResult();

	logger?.LogWarning(task.Exception, "Problem on failing job occured.");
	logger?.LogWarning(task.Exception, "Problem on failing job occurred.");

Implements IJobWorker.IAsyncDisposable #771

Are you sure you want to change the base?

Implements IJobWorker.IAsyncDisposable #771

Uh oh!

Conversation

xlegalles commented Feb 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ChrisKujawa left a comment

Choose a reason for hiding this comment

Uh oh!

ChrisKujawa Feb 10, 2025

Choose a reason for hiding this comment

Uh oh!

xlegalles Feb 10, 2025

Choose a reason for hiding this comment

Uh oh!

ChrisKujawa Feb 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xlegalles Feb 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xlegalles Feb 11, 2025

Choose a reason for hiding this comment

Uh oh!

xlegalles Feb 11, 2025

Choose a reason for hiding this comment

Uh oh!

ChrisKujawa Feb 19, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 7, 2025

Choose a reason for hiding this comment

Uh oh!

ChrisKujawa commented Dec 7, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

xlegalles commented Feb 10, 2025 •

edited

Loading

ChrisKujawa Feb 10, 2025 •

edited

Loading

xlegalles Feb 11, 2025 •

edited

Loading