storage: in remap, when since is empty, suspend instead of panic'ing

[aljoscha]

As the comment describes, this is a race condition that is expected to happen and it's better to suspend rather than bring down the whole cluster, which causes pain for customers/the oncall.

@def- This fixes the panic of incident-360

Motivation

Tips for reviewer

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).
• If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.

[def-]

What would be a good way to test this? Should I try just creating sources, using them, and then deleting them?

[aljoscha]

The case we observed in the incident is the source being dropped right after creating it, I think that's the most promising path. It also helps when the cluster is really busy, so it doesn't try and render the source in time but is delayed to after the source is dropped already

[def-]

I have a simple testdrive test with which I'm trying to reproduce this, but I ran into another panic instead:

thread 'timely:work-0' panicked at src/storage/src/render/sources.rs:223:18:
resuming an already finished ingestion
   5: core::panicking::panic_fmt
   6: core::option::expect_failed
   7: mz_storage::render::sources::render_source::<timely::dataflow::scopes::child::Child<timely::worker::Worker<timely_communication::allocator::generic::Generic>, ()>, mz_storage_types::sources::kafka::KafkaSourceConnection>
   8: mz_storage::render::build_ingestion_dataflow::<timely_communication::allocator::generic::Generic>::{closure#0}::{closure#0}
   9: <timely::worker::Worker<timely_communication::allocator::generic::Generic>>::dataflow_core::<(), (), mz_storage::render::build_ingestion_dataflow<timely_communication::allocator::generic::Generic>::{closure#0}, alloc::boxed::Box<()>>
  10: mz_storage::render::build_ingestion_dataflow::<timely_communication::allocator::generic::Generic>
  11: <mz_storage::storage_state::Worker<timely_communication::allocator::generic::Generic>>::run_client
  12: <mz_storage::server::Config as mz_cluster::types::AsRunnableWorker<mz_storage_client::client::StorageCommand, mz_storage_client::client::StorageResponse>>::build_and_run::<timely_communication::allocator::generic::Generic>
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.

I'll experiment around a bit with making it reproduce more reliably and open a separate PR.