replication: fetch replication log in a loop

[neubaner]

Since batch_log_entries sends up to 1024 frames, the sync operation should fetch all frames until it finds a empty page.

[penberg]

@psarna can you review this?

[psarna]

The issue here is more complicated than switching to log_entries gRPC stream. The log_entries stream is designed for long polls and long-standing connections, ones that continuously fetch new frames from the background and apply them. That's how sqld replicas fetch data. Because of that, log_entries blocks once there are now more frames to replicate, and indefinitely waits for new frames, assuming that they will come. Contrarily, we added batch_log_entries for applications that want to actively poll for new frames and never block. This non-blocking way is especially important for single-threaded apps, e.g. ones that run on WebAssembly - we just ask for new frames from time to time, but never actually block.

The changes in this PR switch to log_entries, but that leaves the blocking problem. Also, note that too-large-batches problem was recently fixed sqld-side, via libsql/sqld#648. There's now a limit of how large a batch of entries can be, and users (e.g. this code) are just expected to ask for more, which effectively implements pagination.

To sum up, I think the proper way to improve libsql code is to modify fn sync_from_http, so that instead of asking for fetch_log_entries once, it does so in a loop, until it hits that 0 pages were returned - and that indicates there are no more frames we can replicate in a non-blocking way.

[neubaner]

@psarna Sorry, I didn't realize that log_entries was a blocking operation. Updated the PR to fetch log entries in a loop until it find a empty page.

[psarna]

The error path gets more complicated with a loop approach: what if we successfully applied 2048 frames, but then failed to replicate the next batch? Current approach just returns an error, but "next_offset" still points after the frames we partially applied. Maybe we should retry a few times, and only return an error if the attempts failed. And if we decide not to retry, which makes the code simpler, there's also another issue: sync_from_http already returns a Result<usize>, and that brings the question what to return if we applied some frames, but failed to apply some more recent ones. We should either return the number of partial frames and report "success", because some frames were applied after all, or create a new specialized error case, like PartialUpdate(usize). I think it's reasonable to return success on partial application, but let's discuss. /cc @penberg @LucioFranco @MarinPostma

Side note - there's a preexisting error here, because we bump next_offset in line 235 before sending frames. If sending fails, we end up with next_offset already updated, and the frames won't be fetched anymore. I'll fix that asap as a separate patch, but meanwhile, the new code should also only update next_offset after frames are successfully sent.
Back to error handling - I think we need to things:

1. Retry if getting frames failed, but some previous frames were already applied in this call - to prevent partial updates.
2. Clearly comment in the code that it's now possible to get a partial result
3. Decide if we return an Ok(usize) on partial application of frames, or some special error code. I think Ok(usize) is fair

[neubaner]

I moved the next_offset update after injector.step() as in the patch you pushed.

The major downside I can see with returning Ok(usize) is that the user of the library would not be able to tell if it was a partial update. That's fine for sync because it runs in a loop anyway, but for sync_oneshot it would be nice to know it the sync was a partial, so the user can retry.

What do you think about returning something like an Ok(ReplicatedFrames) for sync_oneshot instead of an specialized error?

// Could be a struct with a boolean flag or enum
enum ReplicatedFrames {
    Partial(usize),
    Full(usize),
}

impl ReplicatedFrames {
    /// Number of frames that were synced successfully
    pub fn frames_synced(&self) -> usize {
        match self {
            ReplicatedFrames::Partial(frames) => *frames,
            ReplicatedFrames::Full(frames) => *frames,
        }
    }

    /// A partial sync indicates that more frames _might_ be available to fetch
    pub fn sync_partially(&self) -> bool {
        matches!(self, ReplicatedFrames::Partial(_))
    }
}

[penberg]

The pull request has conflicts and has not been updated for a while. Please feel free to open a new one if you wish to submit this for merging.