[Bug] Bounding the file connector's work queue (#28) would silently drop files against current master

[david-streamlio]

Summary

#28 ("Prevent OOM in FileSource by bounding internal queues") is green and does not conflict with anything textually, but merging it as-is against current master would cause silent, permanent file loss. It must not merge without a companion change to FileListingThread.

The interaction

#41 (merged) rewrote FileListingThread to offer each file exactly once while it remains on disk:

for (File f : listing) {
    if (alreadyOffered.add(f)) {
        workQueue.offer(f);      // return value ignored
    }
}

Ignoring the return value was safe because workQueue was an unbounded LinkedBlockingQueue, so offer() always succeeded.

#28 replaces the three queues with bounded ones:

workQueue = new LinkedBlockingQueue<>(queueCapacity);

Now offer() returns false when the queue is full. The file has already been recorded in alreadyOffered, and the prune step only forgets files that have left the disk. So the file is never enqueued, never retried, and never processed — for as long as it exists.

The failure mode is worst precisely under the condition #28 is designed to handle: a backlog large enough to fill the queue.

Reproduction

Against current master, with a bounded workQueue of capacity 1 and 3 files on disk, driving only FileListingThread:

WORKQUEUE_SIZE=1
FILES_ON_DISK=3
AFTER_DRAIN_REOFFERED=0     <-- the other 2 files are never re-offered

After draining the queue (as a consumer would), the lister does not re-offer the two files it dropped. They are lost silently — no exception, no log.

Fix

Only record a file as offered once it has actually been accepted:

for (File f : listing) {
    if (!alreadyOffered.contains(f) && workQueue.offer(f)) {
        alreadyOffered.add(f);
    }
}

A file that cannot be enqueued is simply retried on the next listing pass, which is the natural back-pressure behavior — the lister slows to the consumer's rate instead of discarding work.

The keepFile=true branch already re-checks workQueue.contains(f) each pass and is unaffected.

Ownership

The latent assumption is mine, introduced in #41; #28 merely exposes it. Happy to prepare the FileListingThread fix (with a regression test that bounds the queue and asserts nothing is dropped) so #28 can merge on top of it — say the word and I'll open it.

cc @Praveenkumar76 (author of #28)