Function process_message 

fn process_message(
    response: PeekResponseUnary,
    state: &mut OccState,
    as_of: Timestamp,
    fold_target: Timestamp,
    max_result_size: u64,
    table_desc: &RelationDesc,
) -> ProcessResult

Process one subscribe message, updating state in place.

Data rows are accumulated into state (with per-row constraint and max-result-size checks). Progress messages fold everything below fold_target into the payload and can promote the accumulated diffs to “ready to write”.

fold_target must not exceed the timestamp the next write attempt uses, or the payload takes in a diff that is concurrent with that write.