Merge branch 'main' into bld-debug-flaky-go-tests

codetheweb · codetheweb · commit 6f2e1228f42f · 2025-06-05T13:42:07.000-07:00
diff --git a/.github/actions/tilt-setup-prebuild/action.yaml b/.github/actions/tilt-setup-prebuild/action.yaml
@@ -0,0 +1,20 @@
+name: Tilt Setup & Pre-Build
+description: Set up Tilt prereqs and pre-build/pull images
+runs:
+  using: "composite"
+  steps:
+    - name: Start minikube
+      uses: medyagh/setup-minikube@latest
+      with:
+        driver: none # uses Docker engine on host instead of Docker-in-Docker
+    # tilt ci can automatically build images while bringing up the cluster. However, this results in flaky runs for our usage so we split up image building and pod deployment into separate steps. See https://github.com/chroma-core/chroma/pull/4720.
+    - name: Install Tilt, bake images, pre-pull external images
+      shell: bash
+      env:
+        TILT_VERSION: "0.34.2"
+      run: |
+        parallel --tag --linebuffer ::: \
+          "bash -c 'curl -fsSL https://github.com/tilt-dev/tilt/releases/download/v${TILT_VERSION}/tilt.${TILT_VERSION}.linux.x86_64.tar.gz | tar -xzv -C /usr/local/bin tilt'" \
+          "docker buildx bake -f ${{ github.action_path }}/docker-bake.hcl --load" \
+          "bash ${{ github.action_path }}/pull_external_images.sh"
+      working-directory: ${{ github.action_path }}/../../../ # this allows other repos to reuse this workflow when this repo may not be the current working directory
diff --git a/.github/actions/tilt-setup-prebuild/docker-bake.hcl b/.github/actions/tilt-setup-prebuild/docker-bake.hcl
diff --git a/.github/actions/tilt-setup-prebuild/pull_external_images.sh b/.github/actions/tilt-setup-prebuild/pull_external_images.sh
diff --git a/.github/actions/tilt/action.yaml b/.github/actions/tilt/action.yaml
@@ -3,20 +3,8 @@ description: "This action starts Tilt services"
 runs:
   using: "composite"
   steps:
-    - name: Start minikube
-      uses: medyagh/setup-minikube@latest
-      with:
-        driver: none # uses Docker engine on host instead of Docker-in-Docker
-    # tilt ci can automatically build images while bringing up the cluster. However, this results in flaky runs for our usage so we split up image building and pod deployment into separate steps. See https://github.com/chroma-core/chroma/pull/4720.
-    - name: Install Tilt, bake images, pre-pull external images
-      shell: bash
-      env:
-        TILT_VERSION: "0.34.2"
-      run: |
-        parallel --tag --linebuffer ::: \
-          "bash -c 'curl -fsSL https://github.com/tilt-dev/tilt/releases/download/v${TILT_VERSION}/tilt.${TILT_VERSION}.linux.x86_64.tar.gz | tar -xzv -C /usr/local/bin tilt'" \
-          "docker buildx bake -f .github/actions/tilt/docker-bake.hcl --load" \
-          "bash .github/actions/tilt/pull_external_images.sh"
+    - name: Tilt Setup & Pre-Build
+      uses: ./.github/actions/tilt-setup-prebuild
     - name: Start Tilt
       shell: bash
       run: tilt ci
diff --git a/chromadb/test/utils/wait_for_version_increase.py b/chromadb/test/utils/wait_for_version_increase.py
@@ -23,7 +23,8 @@ def wait_for_version_increase(
     while curr_version == initial_version:
         time.sleep(TIMEOUT_INTERVAL)
         if time.time() - initial_time > timeout:
-            raise TimeoutError("Model was not updated in time")
+            collection_id = client.get_collection(collection_name).id
+            raise TimeoutError(f"Model was not updated in time for {collection_id}")
         curr_version = get_collection_version(client, collection_name)
 
     return curr_version
diff --git a/go/pkg/log/repository/log.go b/go/pkg/log/repository/log.go
@@ -168,6 +168,13 @@ func (r *LogRepository) ForkRecords(ctx context.Context, sourceCollectionID stri
 		}
 	}()
 
+	// NOTE(rescrv):  Only sourceInfo.IsSealed should be used on this struct.
+	var sourceInfo log.Collection
+	sourceInfo, err = queriesWithTx.GetCollection(ctx, sourceCollectionID)
+	if err != nil {
+		sourceInfo.IsSealed = false
+	}
+
 	var sourceBounds log.GetBoundsForCollectionRow
 	sourceBounds, err = queriesWithTx.GetBoundsForCollection(ctx, sourceCollectionID)
 	if err != nil {
@@ -207,6 +214,7 @@ func (r *LogRepository) ForkRecords(ctx context.Context, sourceCollectionID stri
 		ID:                              targetCollectionID,
 		RecordCompactionOffsetPosition:  int64(compactionOffset),
 		RecordEnumerationOffsetPosition: int64(enumerationOffset),
+		IsSealed:                        sourceInfo.IsSealed,
 	})
 	if err != nil {
 		trace_log.Error("Error in updating offset for target collection", zap.Error(err), zap.String("collectionId", targetCollectionID))
diff --git a/go/pkg/log/repository/log_test.go b/go/pkg/log/repository/log_test.go
@@ -30,15 +30,11 @@ func (suite *LogTestSuite) SetupSuite() {
 	connectionString, err := libs2.StartPgContainer(ctx)
 	config.DATABASE_URL = connectionString
 	assert.NoError(suite.t, err, "Failed to start pg container")
-
-	err = libs2.RunMigration(ctx, connectionString)
-	assert.NoError(suite.t, err, "Failed to run migration")
-
-	// Create connection AFTER migration to ensure schema visibility
 	var conn *pgxpool.Pool
 	conn, err = libs2.NewPgConnection(ctx, config)
 	assert.NoError(suite.t, err, "Failed to create new pg connection")
-
+	err = libs2.RunMigration(ctx, connectionString)
+	assert.NoError(suite.t, err, "Failed to run migration")
 	suite.sysDb = sysdb.NewMockSysDB()
 	suite.lr = NewLogRepository(conn, suite.sysDb)
 }
@@ -120,17 +116,17 @@ func (suite *LogTestSuite) TestUniqueConstraintPushLogs() {
 func (suite *LogTestSuite) TestSealedLogWontPush() {
 	ctx := context.Background()
 	collectionId := types.NewUniqueID()
-	params := log.InsertCollectionParams {
-		ID: collectionId.String(),
+	params := log.InsertCollectionParams{
+		ID:                              collectionId.String(),
 		RecordEnumerationOffsetPosition: 1,
-		RecordCompactionOffsetPosition: 0,
+		RecordCompactionOffsetPosition:  0,
 	}
 	_, err := suite.lr.queries.InsertCollection(ctx, params)
 	assert.NoError(suite.t, err, "Initializing log should not fail.")
 	_, err = suite.lr.queries.SealLog(ctx, collectionId.String())
 	assert.NoError(suite.t, err, "Sealing log should not fail.")
 	var isSealed bool
-	_, isSealed, err = suite.lr.InsertRecords(ctx, collectionId.String(), [][]byte{{1,2,3}})
+	_, isSealed, err = suite.lr.InsertRecords(ctx, collectionId.String(), [][]byte{{1, 2, 3}})
 	assert.NoError(suite.t, err, "Failed to push logs")
 	assert.True(suite.t, isSealed, "Did not report was sealed")
 }
diff --git a/go/pkg/log/store/db/queries.sql.go b/go/pkg/log/store/db/queries.sql.go
diff --git a/go/pkg/log/store/queries/queries.sql b/go/pkg/log/store/queries/queries.sql
@@ -4,6 +4,11 @@ FROM collection
 WHERE id = $1
 FOR UPDATE;
 
+-- name: GetCollection :one
+SELECT *
+FROM collection
+WHERE id = $1;
+
 -- name: InsertRecord :copyfrom
 INSERT INTO record_log (collection_id, "offset", record, timestamp) values($1, $2, $3, $4);
 
@@ -30,7 +35,7 @@ UPDATE collection set record_compaction_offset_position = $2 where id = $1;
 UPDATE collection set record_enumeration_offset_position = $2 where id = $1;
 
 -- name: InsertCollection :one
-INSERT INTO collection (id, record_enumeration_offset_position, record_compaction_offset_position) values($1, $2, $3) returning *;
+INSERT INTO collection (id, record_enumeration_offset_position, record_compaction_offset_position, is_sealed) values($1, $2, $3, $4) returning *;
 
 -- name: PurgeRecords :exec
 DELETE FROM record_log r using collection c where r.collection_id = c.id and r.offset <= c.record_compaction_offset_position;
diff --git a/rust/frontend/sample_configs/tilt_config.yaml b/rust/frontend/sample_configs/tilt_config.yaml
@@ -56,3 +56,5 @@ circuit_breaker:
   requests: 1000
 enable_span_indexing: true
 default_knn_index: "spann"
+tenants_to_migrate_immediately:
+- "default_tenant"
diff --git a/rust/log-service/src/lib.rs b/rust/log-service/src/lib.rs
@@ -896,6 +896,14 @@ impl LogServer {
         self.metrics
             .dirty_log_records_read
             .add(dirty_markers.len() as u64, &[]);
+        let Some((last_record_inserted, _)) = dirty_markers.last() else {
+            let backpressure = vec![];
+            self.set_backpressure(&backpressure);
+            let mut need_to_compact = self.need_to_compact.lock();
+            let mut rollups = HashMap::new();
+            std::mem::swap(&mut *need_to_compact, &mut rollups);
+            return Ok(());
+        };
         let mut rollups = DirtyMarker::coalesce_markers(&dirty_markers)?;
         self.enrich_dirty_log(&mut rollups).await?;
         let mut markers = vec![];
@@ -919,7 +927,8 @@ impl LogServer {
             markers.push(serde_json::to_string(&DirtyMarker::Cleared).map(Vec::from)?);
         }
         let mut new_cursor = cursor.clone();
-        new_cursor.position = self.dirty_log.append_many(markers).await?;
+        self.dirty_log.append_many(markers).await?;
+        new_cursor.position = *last_record_inserted + 1u64;
         let Some(cursors) = self.dirty_log.cursors(CursorStoreOptions::default()) else {
             return Err(Error::CouldNotGetDirtyLogCursors);
         };
@@ -1212,23 +1221,11 @@ impl LogServer {
                 max_bytes: None,
                 max_records: Some(pull_logs.batch_size as u64),
             };
-            // NOTE(rescrv):  Log records are immutable, so if a manifest includes our range we can
-            // serve it directly from the scan_from_manifest call.
-            let (manifest_start, manifest_limit) = (
-                manifest.minimum_log_position().offset() as i64,
-                manifest.maximum_log_position().offset() as i64,
-            );
-            if manifest_start <= pull_logs.start_from_offset
-                && pull_logs.start_from_offset + pull_logs.batch_size as i64 <= manifest_limit
-            {
-                LogReader::scan_from_manifest(
-                    &manifest,
-                    LogPosition::from_offset(pull_logs.start_from_offset as u64),
-                    limits,
-                )
-            } else {
-                None
-            }
+            LogReader::scan_from_manifest(
+                &manifest,
+                LogPosition::from_offset(pull_logs.start_from_offset as u64),
+                limits,
+            )
         } else {
             None
         }
@@ -1289,7 +1286,7 @@ impl LogServer {
                     let prefix = storage_prefix_for_log(collection_id);
                     if let Some(cache) = self.cache.as_ref() {
                         let cache_key = format!("{collection_id}::{}", fragment.path);
-                        let cache_span = tracing::info_span!("cache get");
+                        let cache_span = tracing::info_span!("cache get", cache_key = ?cache_key);
                         if let Ok(Some(answer)) = cache.get(&cache_key).instrument(cache_span).await
                         {
                             return Ok(Arc::new(answer.bytes));
diff --git a/rust/tracing/src/init_tracer.rs b/rust/tracing/src/init_tracer.rs
@@ -38,6 +38,7 @@ pub fn init_global_filter_layer() -> Box<dyn Layer<Registry> + Send + Sync> {
                 "wal3",
                 "worker",
                 "garbage_collector",
+                "continuous_verification",
             ]
             .into_iter()
             .map(|s| s.to_string() + "=trace")
@@ -147,6 +148,10 @@ pub fn init_stdout_layer() -> Box<dyn Layer<Registry> + Send + Sync> {
                     .module_path()
                     .unwrap_or("")
                     .starts_with("hosted-frontend")
+                || metadata
+                    .module_path()
+                    .unwrap_or("")
+                    .starts_with("continuous_verification")
         }))
         .with_filter(tracing_subscriber::filter::LevelFilter::INFO)
         .boxed()
diff --git a/rust/wal3/src/reader.rs b/rust/wal3/src/reader.rs
