Network glitch during scatter() may result in memory leak

[crusaderky]

If scatter fails either between these two lines (direct=True)

distributed/distributed/client.py

Lines 2275 to 2281 in 9bb999d

	_, who_has, nbytes = await scatter_to_workers(
	nthreads, data2, report=False, rpc=self.rpc
	)
	await self.scheduler.update_data(
	who_has=who_has, nbytes=nbytes, client=self.id
	)

or these (direct=False)

distributed/distributed/scheduler.py

Lines 4978 to 4982 in 9bb999d

	keys, who_has, nbytes = await scatter_to_workers(
	nthreads, data, rpc=self.rpc, report=False
	)
	self.update_data(who_has=who_has, nbytes=nbytes, client=client)

e.g. the network falls over after the data has reached the worker, but before the worker can respond OK on the RPC channel, then the scheduler will not know about the data and the worker will not inform it.
The data will just sit there consuming memory, unknown to the scheduler, and only a new scatter/compute of the same key on the same worker will fix the issue.

[fjetter]

If the worker disconnects, we're closing it now. Is this problem still possible? I would expect either one of the calls to raise an exception or the key to be transitioned to lost

[crusaderky]

I expect we're not closing the worker on the scheduler if an RPC channel between the client and the worker dies.

I would expect either one of the calls to raise an exception or the key to be transitioned to lost

Precisely, if scatter_to_workers raises an exception, update_data is never called. There's nothing to transition, as the scheduler is completely unaware of the key.