[BUG] OBO between log service and compaction. (#4276)

Compaction assumes that enumeration position t_i means t_i was the last
record seen and therefore next reader should read from t_i + 1.

Log service was built and tested for t_i meaning t_i was the first
record to return.

Note:  I changed the go code, but only by moving a +1 out a layer.  The
inner version was inconsistent with convention, so I updated it.

Author: rescrv

go/pkg/log/repository/log.go

@@ -169,14 +169,18 @@ func (r *LogRepository) GetLastCompactedOffsetForCollection(ctx context.Context,
 	return
 }
 
+// GetBoundsForCollection returns the offset of the last record compacted and the offset of the last
+// record inserted.  Thus, the range of uncompacted records is the interval (start, limit], which is
+// kind of backwards from how it is elsewhere, so pay attention to comments indicating the bias of
+// the offset.
 func (r *LogRepository) GetBoundsForCollection(ctx context.Context, collectionId string) (start, limit int64, err error) {
 	bounds, err := r.queries.GetBoundsForCollection(ctx, collectionId)
 	if err != nil {
 		trace_log.Error("Error in getting minimum and maximum offset for collection", zap.Error(err), zap.String("collectionId", collectionId))
 		return
 	}
 	start = bounds.RecordCompactionOffsetPosition
-	limit = bounds.RecordEnumerationOffsetPosition + 1
+	limit = bounds.RecordEnumerationOffsetPosition
 	err = nil
 	return
 }

go/pkg/log/server/server.go

@@ -55,8 +55,9 @@ func (s *logServer) ScoutLogs(ctx context.Context, req *logservicepb.ScoutLogsRe
 	if err != nil {
 		return
 	}
+	// +1 to convert from the (] bound to a [) bound.
 	res = &logservicepb.ScoutLogsResponse {
-		LimitOffset: int64(limit),
+		FirstUninsertedRecordOffset: int64(limit + 1),
 	}
 	return
 }

idl/chromadb/proto/logservice.proto

@@ -19,7 +19,10 @@ message ScoutLogsRequest {
 }
 
 message ScoutLogsResponse {
-  int64 limit_offset = 1;
+  // This field was once used for an ambiguous last_record_offset alternative.
+  reserved 1;
+  // The next record to insert will have this offset.
+  int64 first_uninserted_record_offset = 2;
 }
 
 message PullLogsRequest {

rust/log-service/src/lib.rs

@@ -671,7 +671,9 @@ impl LogService for LogServer {
                 }
             };
             let limit_offset = limit_position.offset() as i64;
-            Ok(Response::new(ScoutLogsResponse { limit_offset }))
+            Ok(Response::new(ScoutLogsResponse {
+                first_uninserted_record_offset: limit_offset,
+            }))
         }
         .instrument(span)
         .await

rust/log/src/grpc_log.rs

@@ -197,6 +197,7 @@ impl GrpcLog {
         &mut self.client
     }
 
+    // ScoutLogs returns the offset of the next record to be inserted into the log.
     pub(super) async fn scout_logs(
         &mut self,
         collection_id: CollectionUuid,
@@ -216,7 +217,7 @@ impl GrpcLog {
             }
         };
         let scout = response.into_inner();
-        Ok(scout.limit_offset as u64)
+        Ok(scout.first_uninserted_record_offset as u64)
     }
 
     pub(super) async fn read(

rust/log/src/log.rs

@@ -50,6 +50,7 @@ impl Log {
         }
     }
 
+    // ScoutLogs returns the offset of the next record to be inserted into the log.
     #[tracing::instrument(skip(self))]
     pub async fn scout_logs(
         &mut self,

rust/wal3/src/manifest.rs

@@ -644,7 +644,7 @@ impl Manifest {
             },
             (Some(f), None) => f,
             (None, Some(s)) => s,
-            (None, None) => LogPosition::default(),
+            (None, None) => LogPosition::from_offset(1),
         }
     }