fix duplicating&learning(2/n): plog may be lost on going duplicating when replica re-open and then learn

[foreverneverer]

#693 has report one problem, but it hasn't be reproduced in later many test, however, I found this follow problem that can be ususlly reproduced after enough test. so I open the issue and focus fixing this problem

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do?

• dup status is start
• set meta lively to balance after adding three node

2. What did you see instead?

server crash and the core stack:

#0  0x0000003e8d8328a5 in raise () from /lib64/libc.so.6
#1  0x0000003e8d834085 in abort () from /lib64/libc.so.6
#2  0x00007f24bfcf0d4e in dsn_coredump () at /home/jiashuo1/work/pegasus/rdsn/src/runtime/service_api_c.cpp:78
#3  0x00007f24bfac71a3 in dsn::replication::prepare_list::commit (this=this@entry=0x1546d7d90, d=659887, ct=ct@entry=dsn::replication::COMMIT_TO_DECREE_HARD) at /home/jiashuo1/work/pegasus/rdsn/src/replica/prepare_list.cpp:154
#4  0x00007f24bfac7e6a in dsn::replication::prepare_list::prepare (this=0x1546d7d90, mu=..., status=status@entry=dsn::replication::partition_status::PS_INACTIVE, pop_all_committed_mutations=pop_all_committed_mutations@entry=false)
    at /home/jiashuo1/work/pegasus/rdsn/src/replica/prepare_list.cpp:124
#5  0x00007f24bfbf6d93 in dsn::replication::mutation_batch::add (this=0x25cb33bc0, mu=...) at /home/jiashuo1/work/pegasus/rdsn/src/replica/duplication/mutation_batch.cpp:52
#6  0x00007f24bfbf159b in operator() (mu=..., log_bytes_length=2355, __closure=0x7f246d5caaa0) at /home/jiashuo1/work/pegasus/rdsn/src/replica/duplication/load_from_private_log.cpp:124
#7  std::_Function_handler<bool(int, dsn::ref_ptr<dsn::replication::mutation>&), dsn::replication::load_from_private_log::replay_log_block()::<lambda(int, dsn::replication::mutation_ptr&)> >::_M_invoke(const std::_Any_data &, <unknown type in /home/work/app/pegasus/c4tst-tune/replica/package/bin/libdsn_replica_server.so, CU 0x1ff2be0, DIE 0x20e7113>, dsn::ref_ptr<dsn::replication::mutation> &) (__functor=..., __args#0=<optimized out>, __args#1=...)
    at /home/jiashuo1/app/toolchain/gcc540/output/include/c++/5.4.0/functional:1857
#8  0x00007f24bfabb00e in operator() (__args#1=..., __args#0=2355, this=0x7f246d5caaa0) at /home/jiashuo1/app/toolchain/gcc540/output/include/c++/5.4.0/functional:2267
#9  dsn::replication::mutation_log::replay_block(dsn::ref_ptr<dsn::replication::log_file>&, std::function<bool (int, dsn::ref_ptr<dsn::replication::mutation>&)>&, unsigned long, long&) (log=..., callback=..., 
    start_offset=<optimized out>, end_offset=@0x25cb33bb8: 1004176305) at /home/jiashuo1/work/pegasus/rdsn/src/replica/mutation_log_replay.cpp:94
#10 0x00007f24bfbf0b7c in replay_block (end_offset=<optimized out>, start_offset=<optimized out>, 
    callback=<unknown type in /home/work/app/pegasus/c4tst-tune/replica/package/bin/libdsn_replica_server.so, CU 0x1ff2be0, DIE 0x20b8705>, log=...) at /home/jiashuo1/work/pegasus/rdsn/src/replica/mutation_log.h:149
#11 dsn::replication::load_from_private_log::replay_log_block (this=this@entry=0x25cb33b00) at /home/jiashuo1/work/pegasus/rdsn/src/replica/duplication/load_from_private_log.cpp:131
#12 0x00007f24bfbf12ed in dsn::replication::load_from_private_log::run (this=0x25cb33b00) at /home/jiashuo1/work/pegasus/rdsn/src/replica/duplication/load_from_private_log.cpp:68
#13 0x00007f24bfd1bbbd in dsn::task::exec_internal (this=this@entry=0x114855877) at /home/jiashuo1/work/pegasus/rdsn/src/runtime/task/task.cpp:176
#14 0x00007f24bfd3671a in dsn::task_worker::loop (this=0x3040780) at /home/jiashuo1/work/pegasus/rdsn/src/runtime/task/task_worker.cpp:211
#15 0x00007f24bfd368e8 in dsn::task_worker::run_internal (this=0x3040780) at /home/jiashuo1/work/pegasus/rdsn/src/runtime/task/task_worker.cpp:191
#16 0x00007f24bca55fd0 in std::execute_native_thread_routine (__p=<optimized out>) at /home/jiashuo1/app/toolchain/gcc540/objdir/../gcc-5.4.0/libstdc++-v3/src/c++11/thread.cc:84
#17 0x0000003e8dc07851 in start_thread () from /lib64/libpthread.so.0
#18 0x0000003e8d8e811d in clone () from /lib64/libc.so.6

F2021-04-20 17:01:01.121 (1618909261121014136 26dc7) replica.rep_long1.04050011002af702: prepare_list.cpp:155:commit(): assertion expression: mu != nullptr && mu->is_logged()
F2021-04-20 17:01:01.121 (1618909261121033079 26dc7) replica.rep_long1.04050011002af702: prepare_list.cpp:155:commit(): [mutation_batch@210.59@10.132.5.5:32801] mutation 659887 is missing in prepare list

3. What version of Pegasus are you using?

Pegasus 2.1.1