Client story

[sjperkins]

• Closes Client.story - Support collecting cluster-wide story for a key or stimulus ID #5872
• Tests added / passed
• Passes pre-commit run --all-files

[sjperkins]

This PR is based on top of #5954 and should be rebased when #5954 is merged.

TODO:

• At present, assert_story's monotonically increasing timestamp requirement doesn't work well the cluster-wide stories.
• The Client currently requests the Scheduler for story keys, which in turn requests stories from Workers. It may be possible for the Client to simply request stories from the Workers.
• Comm errors when requesting stories are not gracefully handled.

[github-actions]

Unit Test Results

       12 files  ±  0         12 suites  ±0   6h 19m 36s ⏱️ - 16m 38s
  2 673 tests +  4    2 591 ✔️ +  4    81 💤  - 1  1 ❌ +1 
15 942 runs  +24  15 077 ✔️ +22  859 💤  - 4  6 ❌ +6 

For more details on these failures, see this check.

Results for commit 2888b26. ± Comparison against base commit 6dd928b.

♻️ This comment has been updated with latest results.

[sjperkins]

From #5872

The failure behaviour of timed out workers should be configurable (raise, ignore). TBD: Should there be a sentinel if ignored?

and #5987 (comment)

Comm errors when requesting stories are not gracefully handled.

asyncio.gather(..., return_exceptions=True) offers a way forward.

[graingert]

Suggested change

	return super().get_story(*args, **kw)
	return await super().get_story(*args, **kw)

[graingert]

the task will always be done at this point - because of the return await task

[graingert]

I think reproducing the pattern from

distributed/distributed/scheduler.py

Lines 6155 to 6187 in 5c7d555

	async def send_message(addr):
	try:
	comm = await self.rpc.connect(addr)
	comm.name = "Scheduler Broadcast"
	try:
	resp = await send_recv(
	comm, close=True, serializers=serializers, **msg
	)
	finally:
	self.rpc.reuse(addr, comm)
	return resp
	except Exception as e:
	logger.error(f"broadcast to {addr} failed: {e.__class__.__name__}: {e}")
	if on_error == "raise":
	raise
	elif on_error == "return":
	return e
	elif on_error == "return_pickle":
	return dumps(e, protocol=4)
	elif on_error == "ignore":
	return ERROR
	else:
	raise ValueError(
	"on_error must be 'raise', 'return', 'return_pickle', "
	f"or 'ignore'; got {on_error!r}"
	)
	results = await All(
	[send_message(address) for address in addresses if address is not None]
	)
	return {k: v for k, v in zip(workers, results) if v is not ERROR}

would be cleaner than using gather/return_exceptions/cancel as they all interact slightly strangely

[fjetter]

Closed by #6161