PartitionAvailabilityTracker self-retains after the final producer owner is dropped

[helenalgebra]

Summary

PartitionAvailabilityTracker::start spawns a detached task that captures a strong Arc to the tracker. The tracker’s Drop implementation signals that task to terminate, but the task-held Arc prevents Drop from running after the final TopicProducer owner is dropped.

The resulting ownership cycle is:

1. start constructs an Arc<PartitionAvailabilityTracker>.
2. The spawned task captures a clone as shared_tracker.
3. The task exits only when shared_tracker.terminate is notified.
4. Drop for PartitionAvailabilityTracker is the only code that notifies terminate.
5. The task-held Arc prevents that Drop during normal executor operation.

Current source:
https://github.com/fluvio-community/fluvio/blob/master/crates/fluvio/src/producer/mod.rs

Impact

Each independently constructed TopicProducer starts an availability tracker. Dropping all handles to that producer does not release the tracker or stop its task.

The retained task is active rather than merely allocated: it continues waking once per second and querying the topic and partition stores. Applications that repeatedly construct and retire producers can therefore accumulate background tasks and retained tracker state for the lifetime of the executor.

Clones of one TopicProducer share the same tracker, so this concerns independently constructed producers rather than individual cloned handles.

Suggested invariant

After the final externally reachable owner of a producer is dropped, its availability tracker must stop retaining itself and eventually become unreachable.

A focused unit regression could:

1. Start a PartitionAvailabilityTracker.
2. Retain a Weak reference to it.
3. Drop the returned external Arc.
4. Drive the executor through the task’s next scheduling point.
5. Assert, with a bounded timeout, that Weak::upgrade() returns None.

Repeating this sequence would also detect task accumulation across producer construction and retirement.

Possible fix direction

The task should not own the object whose Drop is responsible for stopping that task. Two possible approaches are:

• Capture a Weak<PartitionAvailabilityTracker> and hold any upgraded strong reference only for a bounded update operation, releasing it before sleeping.
• Separate task state from the externally owned shutdown guard, so dropping the guard can signal termination without depending on destruction of task-owned state.

Either approach should preserve the current partition-refresh behavior while allowing normal producer ownership to determine the tracker’s lifetime.