[BUG] InMemoryLog preserve offset orders

rust/worker/src/execution/operators/pull_log.rs

@@ -116,7 +116,7 @@ impl Operator<PullLogsInput, PullLogsOutput> for PullLogsOperator {
             }
 
             num_records_read += logs.len();
-            offset += batch_size as i64;
+            offset = logs.last().unwrap().log_offset + 1;
             result.append(&mut logs);
 
             // We used a a timestamp and we didn't get a full batch, so we have retrieved

rust/worker/src/log/log.rs

@@ -261,7 +261,17 @@ impl InMemoryLog {
 
     pub fn add_log(&mut self, collection_id: String, log: Box<InternalLogRecord>) {
         let logs = self.logs.entry(collection_id).or_insert(Vec::new());
-        logs.push(log);
+        if logs.is_empty() {
+            logs.push(log);
+            return;
+        } else {
+            let last_log = logs.last().unwrap();
+            if log.log_offset != last_log.log_offset + 1 {
+                panic!("log offset is not in order");
+            } else {
+                logs.push(log);
+            }
+        }
     }
 }
 
@@ -284,11 +294,17 @@ impl Log for InMemoryLog {
             None => return Ok(Vec::new()),
         };
         let mut result = Vec::new();
-        for i in offset..(offset + batch_size as i64) {
-            if i < logs.len() as i64 && logs[i as usize].log_ts <= end_timestamp {
-                result.push(logs[i as usize].record.clone());
+        for i in 0..logs.len() as i64 {
+            if logs[i as usize].log_offset < offset {
+                continue;
+            }
+            if logs[i as usize].log_ts > end_timestamp {
+                continue;
             }
+            result.push(logs[i as usize].record.clone());
         }
+        result.sort_by(|a, b| a.log_offset.cmp(&b.log_offset));
+        result.truncate(batch_size as usize);
         Ok(result)
     }