rgw/sfs: mark all OPEN versions DELETED on startup

src/rgw/driver/sfs/sqlite/sqlite_versioned_objects.cc

@@ -538,4 +538,19 @@ SQLiteVersionedObjects::remove_deleted_versions_transact(uint max_objects
   return retry.run();
 }
 
+int SQLiteVersionedObjects::set_all_open_versions_to_deleted() const {
+  // This function is only for use when we want to deliberately garbage
+  // collect open versions on startup.
+  auto storage = conn->get_storage();
+  auto transaction = storage.transaction_guard();
+  transaction.commit_on_destroy = true;
+  auto now = ceph::real_clock::now();
+  storage.update_all(
+      set(c(&DBVersionedObject::delete_time) = now,
+          c(&DBVersionedObject::object_state) = ObjectState::DELETED),
+      where(is_equal(&DBVersionedObject::object_state, ObjectState::OPEN))
+  );
+  return storage.changes();
+}
+
 }  // namespace rgw::sal::sfs::sqlite

src/rgw/driver/sfs/sqlite/sqlite_versioned_objects.h

@@ -80,6 +80,8 @@ class SQLiteVersionedObjects {
       uint max_objects
   ) const;
 
+  int set_all_open_versions_to_deleted() const;
+
  private:
   std::optional<DBVersionedObject>
   get_committed_versioned_object_specific_version(

src/rgw/rgw_sal_sfs.cc

@@ -607,6 +607,10 @@ SFStore::SFStore(CephContext* c, const std::filesystem::path& data_path)
       ) {
   maybe_init_store();
   db_conn = std::make_shared<sfs::sqlite::DBConn>(cctx);
+  sfs::sqlite::SQLiteVersionedObjects objs_versions(db_conn);
+  int num_deleted = objs_versions.set_all_open_versions_to_deleted();
+  ldout(ctx(), 10) << "marked " << num_deleted << " open objects deleted"
+                   << dendl;
   gc = std::make_shared<sfs::SFSGC>(cctx, this);
 
   filesystem_stats_updater = make_named_thread(

src/test/rgw/sfs/test_rgw_sfs_sqlite_versioned_objects.cc

@@ -1835,3 +1835,50 @@ TEST_F(TestSFSSQLiteVersionedObjects, TestDeleteMarkerNotAlwaysOnTop) {
   );
   EXPECT_EQ(4, versions[4].id);
 }
+
+TEST_F(TestSFSSQLiteVersionedObjects, TestSetAllOpenVersionsToDeleted) {
+  auto ceph_context = std::make_shared<CephContext>(CEPH_ENTITY_TYPE_CLIENT);
+  ceph_context->_conf.set_val("rgw_sfs_data_path", getTestDir());
+  ceph_context->_log->start();
+
+  EXPECT_FALSE(fs::exists(getDBFullPath()));
+  DBConnRef conn = std::make_shared<DBConn>(ceph_context.get());
+
+  auto db_versioned_objects = std::make_shared<SQLiteVersionedObjects>(conn);
+
+  // Create the object, we need it because of foreign key constrains
+  createObject(
+      TEST_USERNAME, TEST_BUCKET, TEST_OBJECT_ID, ceph_context.get(), conn
+  );
+
+  auto object = createTestVersionedObject(1, TEST_OBJECT_ID, "1");
+  db_versioned_objects->insert_versioned_object(object);
+  object.version_id = "test_version_id_2";
+  object.object_state = rgw::sal::sfs::ObjectState::COMMITTED;
+  db_versioned_objects->insert_versioned_object(object);
+  object.version_id = "test_version_id_3";
+  object.object_state = rgw::sal::sfs::ObjectState::DELETED;
+  db_versioned_objects->insert_versioned_object(object);
+  // We now have three versions, one open, one committed, and one deleted.
+  // Setting all open versions to deleted should only impact one row.
+  EXPECT_EQ(db_versioned_objects->set_all_open_versions_to_deleted(), 1);
+
+  uuid_d uuid;
+  uuid.parse(TEST_OBJECT_ID.c_str());
+  auto versions = db_versioned_objects->get_versioned_objects(uuid, false);
+  ASSERT_EQ(3, versions.size());
+
+  // And now we should have two deleted versions and one committed version
+  int committed = 0;
+  int deleted = 0;
+  for (auto v : versions) {
+    if (v.object_state == rgw::sal::sfs::ObjectState::COMMITTED) {
+      ++committed;
+    }
+    if (v.object_state == rgw::sal::sfs::ObjectState::DELETED) {
+      ++deleted;
+    }
+  }
+  EXPECT_EQ(1, committed);
+  EXPECT_EQ(2, deleted);
+}