Skip to content

Fix flaky CI: compaction filter race and iterator use-after-free #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
benoitc merged 2 commits into from
Jun 14, 2026
Merged

Fix flaky CI: compaction filter race and iterator use-after-free #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
59afe70
Harden compaction filter tests against compaction race
benoitc Jun 14, 2026
d74f88c
Fix use-after-free in column-family and transaction iterators
benoitc Jun 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion c_src/erocksdb_iter.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,10 +165,10 @@ Iterator(

ItrObject * itr_ptr;
rocksdb::Iterator * iterator;
ReferencePtr<ColumnFamilyObject> cf_ptr;

if(argc==3)
{
ReferencePtr<ColumnFamilyObject> cf_ptr;
if(!enif_get_cf(env, argv[1], &cf_ptr))
{
delete opts;
Expand All @@ -185,6 +185,11 @@ Iterator(

itr_ptr = ItrObject::CreateItrObject(db_ptr.get(), itr_env, iterator);

// pin the column family so GC of its handle cannot free it while this
// iterator is still using it
if(cf_ptr.get() != nullptr)
itr_ptr->KeepResource(cf_ptr.get());

if(bounds.upper_bound_slice != nullptr)
{
itr_ptr->SetUpperBoundSlice(bounds.upper_bound_slice);
Expand Down
16 changes: 16 additions & 0 deletions c_src/refobjects.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -859,6 +859,15 @@ ItrObject::SetLowerBoundSlice(rocksdb::Slice *slice) {
}


void
ItrObject::KeepResource(void *Resource) {
if (nullptr != Resource) {
enif_keep_resource(Resource);
m_KeptResources.push_back(Resource);
}
}



ItrObject::ItrObject(
DbObject *DbPtr,
Expand Down Expand Up @@ -896,6 +905,13 @@ ItrObject::~ItrObject() {
delete m_Iterator;
//m_Iterator = nullptr;

// release any resources (column family, transaction) pinned for this
// iterator's lifetime. Done after deleting the iterator, which still
// references them.
for (void *resource : m_KeptResources)
enif_release_resource(resource);
m_KeptResources.clear();

return;

} // ItrObject::~ItrObject
Expand Down
9 changes: 9 additions & 0 deletions c_src/refobjects.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,11 @@ class ItrObject : public ErlRefObject
rocksdb::Slice *upper_bound_slice;
rocksdb::Slice *lower_bound_slice;

//!< erlang resources (column family, transaction) kept alive with
//!< enif_keep_resource so they cannot be garbage collected and freed
//!< while this iterator still references them. Released in ~ItrObject.
std::list<void *> m_KeptResources;

protected:
static ErlNifResourceType* m_Itr_RESOURCE;

Expand All @@ -354,6 +359,10 @@ class ItrObject : public ErlRefObject
void SetUpperBoundSlice(rocksdb::Slice*);

void SetLowerBoundSlice(rocksdb::Slice*);

//!< Pin an erlang resource (e.g. column family, transaction) for this
//!< iterator's lifetime so GC cannot free it while the iterator uses it.
void KeepResource(void * Resource);
}; // class ItrObject

/**
Expand Down
9 changes: 8 additions & 1 deletion c_src/transaction.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,9 +481,9 @@ namespace erocksdb {

ItrObject * itr_ptr;
rocksdb::Iterator * iterator;
ReferencePtr<ColumnFamilyObject> cf_ptr;

if (argc == 3) {
ReferencePtr<ColumnFamilyObject> cf_ptr;
if(!enif_get_cf(env, argv[1], &cf_ptr)) {
delete bounds.upper_bound_slice;
delete bounds.lower_bound_slice;
Expand All @@ -497,6 +497,13 @@ namespace erocksdb {

itr_ptr = ItrObject::CreateItrObject(tx_ptr->m_DbPtr.get(), itr_env, iterator);

// pin the transaction (its write batch backs the iterator's delta) and
// the column family so neither can be garbage collected and freed while
// this iterator is still using them
itr_ptr->KeepResource(tx_ptr.get());
if(cf_ptr.get() != nullptr)
itr_ptr->KeepResource(cf_ptr.get());

if(bounds.upper_bound_slice != nullptr) {
itr_ptr->SetUpperBoundSlice(bounds.upper_bound_slice);
}
Expand Down
24 changes: 16 additions & 8 deletions test/compaction_filter.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,8 @@ filter_key_prefix_test() ->

%% Flush and compact with force to ensure filter runs on all data
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"keep_key">>, <<"tmp_key~">>, [{bottommost_level_compaction, force}]),

%% Check results - tmp_ keys should be deleted
TmpResult = rocksdb:get(Db, <<"tmp_key50">>, []),
Expand Down Expand Up @@ -96,7 +97,8 @@ filter_key_suffix_test() ->

%% Flush and compact with force to ensure filter runs on all data
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"key">>, <<"key~">>, [{bottommost_level_compaction, force}]),

%% Verify active keys are kept
{ok, _} = rocksdb:get(Db, <<"key50_active">>, []),
Expand Down Expand Up @@ -196,7 +198,8 @@ filter_multiple_rules_test() ->

%% Flush and compact with force to ensure filter runs on all data
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"keep">>, <<"tmp~">>, [{bottommost_level_compaction, force}]),

%% Verify kept keys are still there
{ok, _} = rocksdb:get(Db, <<"keep_key50">>, []),
Expand Down Expand Up @@ -240,7 +243,8 @@ filter_ttl_from_key_test() ->

%% Flush and force compaction
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<ExpiredTs:64/big>>, <<ValidTs:64/big, 255>>, [{bottommost_level_compaction, force}]),

%% ASSERT: Valid keys should remain
ValidKey1 = <<ValidTs:64/big, "valid_data1">>,
Expand Down Expand Up @@ -294,7 +298,8 @@ filter_erlang_handler_test() ->

%% Flush and force compaction
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"delete">>, <<"keep~">>, [{bottommost_level_compaction, force}]),

%% Wait for handler to process and collect total processed count
TotalProcessed = collect_handler_processed(0, 2000),
Expand Down Expand Up @@ -451,7 +456,8 @@ filter_handler_change_value_test() ->

%% Flush and force compaction
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"normal">>, <<"transform~">>, [{bottommost_level_compaction, force}]),

%% Wait for handler to process
TotalProcessed = collect_handler_processed(0, 2000),
Expand Down Expand Up @@ -588,7 +594,8 @@ filter_key_contains_test() ->

%% Flush and force compaction
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"user">>, <<"user~">>, [{bottommost_level_compaction, force}]),

%% ASSERT: keys containing _session_ should be deleted
not_found = rocksdb:get(Db, <<"user_session_50_data">>, []),
Expand Down Expand Up @@ -630,7 +637,8 @@ filter_value_prefix_test() ->

%% Flush and force compaction
ok = rocksdb:flush(Db, []),
ok = rocksdb:compact_range(Db, undefined, undefined, [{bottommost_level_compaction, force}]),
timer:sleep(100),
ok = rocksdb:compact_range(Db, <<"marked">>, <<"normal~">>, [{bottommost_level_compaction, force}]),

%% ASSERT: keys with DELETED: value prefix should be removed
not_found = rocksdb:get(Db, <<"marked_key50">>, []),
Expand Down
Loading