rgw/sfs: mark all OPEN versions DELETED on startup

All OPEN versions that exist on startup are due to aborted uploads.
Setting them to DELETED allows us to free up space leveraging the GC.

Tested by running `s3cmd --limit-rate 1000 put obj.1mb.bin s3://foo/slow`
and hitting CTRL-C after a few seconds to leave an open object lying
around, then restarting s3gw and looking for "marked 0 open objects deleted"
in the debug logs.

Fixes: https://github.com/aquarist-labs/s3gw/issues/624
Signed-off-by: Tim Serong <[email protected]>

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

@@ -538,4 +538,20 @@ 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::mtime) = 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);
+}