[sinks] Generate updated as_of in rehydrating client

[cjubb39]

Previously, we always used the same as_of when the storaged instance was rehydrated. This meant that the following sequence would result in a panic:

• CREATE SINK FROM SOURCE s (which has as_of t_source)
• We create the sink with AS OF now() = t_sink > t_source
• the persistent collection for source s is compacted so that t_source_1 > t_sink`
• the storaged process crashes and is rehydrated
• we recreate the sink using the cached command that uses t_sink
• This panics when trying to read from the persist collection for the source s

We handled the case when the source was recreated on restart of environmentd by looking at the current since of the source in the storage controller. However, in this case, the storage controller doesn't instruct the client to restart -- the rehydrating task does. So we simply move the logic down a layer

Motivation

Fixes #14555

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered.

• This PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way) and therefore is tagged with a T-protobuf label.

• This PR includes the following user-facing behavior changes:

[cjubb39]

This is ready for review! I was not able to reliably reproduce the panic on main -- but I also haven't been able to reproduce it with this branch. That being said: we are pretty convinced -- just by inspection -- that the current code is wrong.

[aljoscha]

I think this should work. But I had a concern about race conditions. Might be that we have to merge as is.

In any case, this was some nice sleuthing! 😊

[aljoscha]

Now that I look at this I'm a bit concerned about races. The following might happen:

1. controller holds a read handle for the implied since t
2. the actual since of the shard is t - 10 because someone else is still holding a handle
3. we set the as_of to t - 10 because of this
4. that third party released their hold, shard since advances to t
5. the sink, in storaged tries to read and fails

I think 1. and 2. can't currently happen together, because we know that the controller initializes the implied frontier to the shard since. But we might change that in the future, maybe by accident.

It would be nicer if we can thread through the exact since that the controller is holding, but that might not be easily feasible. In that case we should probably merge as is. Tricky ... 🙈

[cjubb39]

I think the correctness here is slightly more embedded than you describe: one of the first things the storage controller does is install a dependency between the from source / table / etc and this newly created sink. That will keep it from updating the read capability of the source

so, while someone else can definitely downgrade their handle, the handle managed by the storage controller should keep the since for the collection itself from being downgraded

[aljoscha]

The rehydration logic is also used when the controller restarts? I'm asking because we removed the logic from there.

[cjubb39]

it is, that's correct!