In #5480 we found that when Worker.batched_stream is restarted after a broken connection to the scheduler, it enters a broken state where send succeeds, but doesn't actually send data (it just sits in the buffer forever).
This is probably fixed by #5457, but I think it may deserve a more thorough fix. If we want BatchedSend to be restartable, it should have a clear interface and tests for this.
xref #4133 #4163 #5377