Scale in unregistered htex blocks

[benclifford]

This PR will make the htex scale-in code consider blocks which have been launched through the provider mechanism but have not registered to the interchange. This is especially relevant at shutdown time, when previous blocks still waiting in the queue (and so, not registered) would be abandoned by parsl shutdown.

[first commit on this PR is a test that should break; when that happens, I'll push the rest; this is to get a demonstrated failure of test in CI]

Changed Behaviour

More blocks scaling in. Tidier shutdown.

Fixes

Fixes #2627

Type of change

• Bug fix

[benclifford]

Since I wrote this code a few weeks ago, there's been a change to the high throughput executor status method to introduce a "MISSING" state, and I've also gone over the general status code with @yadudoc a bit.

Most importantly, I'd like to check that this code isn't actually calling the provider status calls (eg. invoking sbatch) unnecessarily: the current code is quite unclear as to when asking for status returns some in-memory cache of blocks (and block status) vs calling out to the LRM.

There might also be opportunities for aligning this code better with the code that handles MISSING status.

[yadudoc]

This draft PR fixes the issue described here: #3235.

[benclifford]

@yadudoc - I don't think it does? this PR doesn't fix any cache-based status update delays? I thought we tested a different (no-PR-number) patch?

[khk-globus]

Makes the earlier assignment redundant, yeah?

[khk-globus]

provider_blocks is a dictionary, yeah? If so, then b_id is guaranteed to be unique through the loop. Coupled with the fact that block_info is empty when the loop starts, the if-conditional is a waste. Could merge it all into the instantiation:

block_info = {
  block_id: BlockInfo(tasks=0, idle=float('inf')
  for block_id, job_status in provider_blocks.items()
  if job_status.state not in TERMINAL_STATES
}

Later on, block_info keys are set to zero on first use as well, so could make that structure a defaultdict:

block_info: Dict[str, BlockInfo = defaultdict(lambda: BlockInfo(0, float('inf')))
for b_id, job_status in provider_blocks.items():
    # skip blocks that have already finished...
    if job_status.state in TERMINAL_STATES:
        continue

    # the defaultdict logic is now the authority on "empty BlockInfo" objects
    block_info[b_id]

And could remove the if-branch in the following for-loop as well.