bug(server): fix lns mismatch in replication #4967
Conversation
adiholden
changed the title
@romange I still see failure in the lsn in CI with this fix. I will continue investigate this |
Signed-off-by: adi_holden <adi@dragonflydb.io>
Signed-off-by: adi_holden <adi@dragonflydb.io>
Signed-off-by: adi_holden <adi@dragonflydb.io>
|
I suspect the problem relies in JournalStreamer register callback meaning that from a transaction flow if we need to throttle we preempt before writing the data to socket |
|
Lets take it offline - I need to understand more |
Signed-off-by: adi_holden <adi@dragonflydb.io>
|
I got the check fail on master for the lsn order validation. This proves the theory fyi @romange |
Signed-off-by: adi_holden <adi@dragonflydb.io>
|
Test failed again. I suspect that moving the throttle to after the write did not fix the issue because we can have few registered callbacks and throttle on the first one can result in journal reorder. |
|
What if we do not call ThrottleIfNeeded there? what if instead of storing callbacks in change_cb_arr_, we will store pointers to JournalConsumer interfaces that will have: And then |
| @@ -149,11 +149,19 @@ cv_status Transaction::BatonBarrier::Wait(time_point tp) { | |||
|
|
|||
| Transaction::Guard::Guard(Transaction* tx) : tx(tx) { | |||
| DCHECK(tx->cid_->opt_mask() & CO::GLOBAL_TRANS); | |||
| tx->Execute([](auto*, auto*) { return OpStatus::OK; }, false); | |||
| auto cb = [&](Transaction* t, EngineShard* shard) { | |||
| namespaces->GetDefaultNamespace().GetDbSlice(shard->shard_id()).SetExpireAllowed(false); | |||
There was a problem hiding this comment.
we check the same flag in eviction flow FreeMemWithEvictionStep checks expire_allowed_
Signed-off-by: adi_holden <adi@dragonflydb.io>
src/server/journal/journal_slice.cc
Outdated
| @@ -157,28 +157,35 @@ void JournalSlice::CallOnChange(const JournalItem& item) { | |||
| // CallOnChange is atomic iff JournalSlice::SetFlushMode(false) is called before. | |||
| std::shared_lock lk(cb_mu_); | |||
|
|
|||
| const size_t size = change_cb_arr_.size(); | |||
| auto k_v = change_cb_arr_.begin(); | |||
| const size_t size = journal_consumers_arr_.size(); | |||
There was a problem hiding this comment.
the first loop is atomic -let's just iterate over map.
src/server/journal/journal_slice.cc
Outdated
| ++k_v; | ||
| } | ||
| k_v = journal_consumers_arr_.begin(); | ||
| if (enable_journal_flush_) { | ||
| for (size_t i = 0; i < size; ++i) { |
There was a problem hiding this comment.
move size initialization here and add a comment that you set size because ThrottleIfNeeded can preempt and meanwhile journal_consumers_arr_ can grow. and btw, calling ThrottleIfNeeded on a new consumer is harmless so now you can remove size workaround at all, imho
Signed-off-by: adi_holden <adi@dragonflydb.io>
The bug:
entries were expired at the time of transitioning from full sync to stable sync leading to journal changes not propagated to replica.
The fix:
disable expire items on transitioning from full sync to stable sync using