[ENH] Add a lower bound to the scout logs response.

This allows one scout-logs request to know everything that's on the log.
It's necessary for the range to be everything pull-logs can access.

Author: rescrv

go/pkg/log/server/server.go

@@ -54,16 +54,18 @@ func (s *logServer) ScoutLogs(ctx context.Context, req *logservicepb.ScoutLogsRe
 	if err != nil {
 		return
 	}
+	var start int64
 	var limit int64
-	_, limit, err = s.lr.GetBoundsForCollection(ctx, collectionID.String())
+	start, limit, err = s.lr.GetBoundsForCollection(ctx, collectionID.String())
 	if err != nil {
 		return
 	}
 	// +1 to convert from the (] bound to a [) bound.
 	res = &logservicepb.ScoutLogsResponse{
+		FirstUncompactedRecordOffset: int64(start + 1),
 		FirstUninsertedRecordOffset: int64(limit + 1),
 	}
-	trace_log.Info("Scouted Logs", zap.Int64("limit", int64(limit + 1)), zap.String("collectionId", req.CollectionId))
+	trace_log.Info("Scouted Logs", zap.Int64("start", int64(start + 1)), zap.Int64("limit", int64(limit + 1)), zap.String("collectionId", req.CollectionId))
 	return
 }
 

idl/chromadb/proto/logservice.proto

@@ -24,6 +24,8 @@ message ScoutLogsResponse {
   reserved 1;
   // The next record to insert will have this offset.
   int64 first_uninserted_record_offset = 2;
+  // The oldest record on the log will have this offset.
+  int64 first_uncompacted_record_offset = 3;
 }
 
 message PullLogsRequest {

rust/log-service/src/lib.rs

@@ -716,18 +716,24 @@ impl LogService for LogServer {
                 Arc::clone(&self.storage),
                 prefix,
             );
-            let limit_position = match log_reader.maximum_log_position().await {
-                Ok(limit_position) => limit_position,
+            let (start_position, limit_position) = match log_reader.manifest().await {
+                Ok(Some(manifest)) => (
+                    manifest.minimum_log_position(),
+                    manifest.maximum_log_position(),
+                ),
+                Ok(None) => (LogPosition::from_offset(0), LogPosition::from_offset(1)),
                 Err(err) => {
                     if err.code() == chroma_error::ErrorCodes::FailedPrecondition {
-                        LogPosition::from_offset(1)
+                        (LogPosition::from_offset(1), LogPosition::from_offset(1))
                     } else {
                         return Err(Status::new(err.code().into(), err.to_string()));
                     }
                 }
             };
+            let start_offset = start_position.offset() as i64;
             let limit_offset = limit_position.offset() as i64;
             Ok(Response::new(ScoutLogsResponse {
+                first_uncompacted_record_offset: start_offset,
                 first_uninserted_record_offset: limit_offset,
             }))
         }

rust/wal3/src/manifest.rs

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