Function apply_command 

async fn apply_command(
    sink_id: GlobalId,
    corrections: OkErr<Correction<Row>, Correction<DataflowErrorSer>>,
    writer: &WriteHandle<SourceData, (), Timestamp, StorageDiff>,
    cmd: WriteCommand,
    resp_tx: &UnboundedSender<WriteResponse>,
) -> OkErr<Correction<Row>, Correction<DataflowErrorSer>>

Apply a single command to the task state, returning the correction buffers.

desired updates enter corrections as positive contributions and persist updates as negative contributions, so the buffer contains desired - persist, i.e. the updates that need to be written to bring the shard in line with desired.

Correction maintenance is CPU-bound and unbounded in duration: an insert can merge chains spanning the whole buffer and a consolidation sorts it, neither with an await point in between. Running that inline occupies a Tokio worker thread for the entire time, which stops it from polling every other task scheduled on it. It therefore runs on a blocking thread, which the OS can preempt. The buffers move into the blocking closure and back out again because this task owns them exclusively.