Bug

**Describe the bug**
I am trying to do some testing around the message persistence and I am running into a few unexpected behaviors. I want to make sure my expectations of the feature and what you offer are in line. The biggest thing I am noticing is missing messages when I would expect persistence due to loss of connection and client instability. I have included a copy of my config file at the bottom. 

**Expected behavior**
I would expect that despite network loss and client instability, all messages would still be received due to data persistence.

**Actual Behavior**
We are seeing lost messages.

**To Reproduce**
The test is sending 100 messages from Device A to Device B and manipulating connectivity issues to test message persistence.  
Device A establishes a broker bridge to EMQX on a server, and Device B is connected directly to EMQX. We slowly start sending 100 messages. After a little time (10-15 messages) we disconnect Device A’s connection to the network. After some more time, we are stopping the broker client on Device A. After some time, we re-established the network and also started the broker up. After that, we validate how many messages we received in total.

**Environment Details**
Broker to Broker
Go client
Debian Environment 

**Client SDK**
nanomq-sqlite

**Additional context**
nanomq.conf

> mqtt.property_size = 32
> mqtt.max_packet_size = 10KB
> mqtt.max_mqueue_len = 2048
> mqtt.retry_interval = 10s
> mqtt.keepalive_multiplier = 1.25
> mqtt.max_inflight_window = 2048
> mqtt.max_awaiting_rel = 10s
> mqtt.await_rel_timeout = 10s
> 
> system.parallel = 1024
> 
> listeners.tcp.bind="0.0.0.0:1883"
> listeners.ws.bind="0.0.0.0:8083/mqtt"
> 
> bridges.mqtt.emqx1.server="*path to our server*"
> bridges.mqtt.emqx1.proto_ver=5
> bridges.mqtt.emqx1.clientid="7e18aa9a-ed43-11ee-a8e2-0662bd2efbf9"
> bridges.mqtt.emqx1.keepalive=60s
> bridges.mqtt.emqx1.clean_start=false
> bridges.mqtt.emqx1.ssl.keyfile="*path to our keyfile*"
> bridges.mqtt.emqx1.ssl.certfile="*path to our cert file*"
> bridges.mqtt.emqx1.ssl.cacertfile="*path to our ca cert*"
> bridges.mqtt.emqx1.conn_properties.receive_maximum=10
> bridges.mqtt.emqx1.conn_properties.session_expiry_interval=86400
> bridges.mqtt.emqx1.forwards.1.local_topic="test"
> bridges.mqtt.emqx1.forwards.1.remote_topic="7e18aa9a-ed43-11ee-a8e2-0662bd2efbf9/test"
> bridges.mqtt.emqx1.subscription.1.local_topic="test"
> bridges.mqtt.emqx1.subscription.1.remote_topic="7e18aa9a-ed43-11ee-a8e2-0662bd2efbf9/test"
> bridges.mqtt.emqx1.subscription.1.qos=2
> 
> bridges.mqtt.cache.disk_cache_size=102400
> bridges.mqtt.cache.mounted_file_path="/var/nanomq"
> bridges.mqtt.cache.flush_mem_threshold=1
> bridges.mqtt.cache.resend_interval=3000
> 
> http_server.port = 8081
> http_server.limit_conn = 2
> http_server.username = admin
> http_server.password = public
> http_server.auth_type = basic
> #http_server.jwt.public.keyfile = "/etc/certs/jwt/jwtRS256.key.pub"
> 
> log.to = [file, console]
> log.level = warn
> log.dir = "/var/log"
> log.file = "nanomq.log"
> log.rotation.size = 10MB
> log.rotation.count = 5
> 
> auth.allow_anonymous = true
> auth.no_match = allow
> auth.deny_action = ignore
> auth.cache.max_size = 32
> auth.cache.ttl = 1m

Unexpected behavior in data persistence
This is a GitHub Issue
repo:nanomq
owner:emqx
Title : Unexpected behavior in data persistence
Issue date:

--- start body ---
Bug

**Describe the bug**
I am trying to do some testing around the message persistence and I am running into a few unexpected behaviors. I want to make sure my expectations of the feature and what you offer are in line. The biggest thing I am noticing is missing messages when I would expect persistence due to loss of connection and client instability. I have included a copy of my config file at the bottom. 

**Expected behavior**
I would expect that despite network loss and client instability, all messages would still be received due to data persistence.

**Actual Behavior**
We are seeing lost messages.

**To Reproduce**
The test is sending 100 messages from Device A to Device B and manipulating connectivity issues to test message persistence.  
Device A establishes a broker bridge to EMQX on a server, and Device B is connected directly to EMQX. We slowly start sending 100 messages. After a little time (10-15 messages) we disconnect Device A’s connection to the network. After some more time, we are stopping the broker client on Device A. After some time, we re-established the network and also started the broker up. After that, we validate how many messages we received in total.

**Environment Details**
Broker to Broker
Go client
Debian Environment 

**Client SDK**
nanomq-sqlite

**Additional context**
nanomq.conf

> mqtt.property_size = 32
> mqtt.max_packet_size = 10KB
> mqtt.max_mqueue_len = 2048
> mqtt.retry_interval = 10s
> mqtt.keepalive_multiplier = 1.25
> mqtt.max_inflight_window = 2048
> mqtt.max_awaiting_rel = 10s
> mqtt.await_rel_timeout = 10s
> 
> system.parallel = 1024
> 
> listeners.tcp.bind="0.0.0.0:1883"
> listeners.ws.bind="0.0.0.0:8083/mqtt"
> 
> bridges.mqtt.emqx1.server="*path to our server*"
> bridges.mqtt.emqx1.proto_ver=5
> bridges.mqtt.emqx1.clientid="7e18aa9a-ed43-11ee-a8e2-0662bd2efbf9"
> bridges.mqtt.emqx1.keepalive=60s
> bridges.mqtt.emqx1.clean_start=false
> bridges.mqtt.emqx1.ssl.keyfile="*path to our keyfile*"
> bridges.mqtt.emqx1.ssl.certfile="*path to our cert file*"
> bridges.mqtt.emqx1.ssl.cacertfile="*path to our ca cert*"
> bridges.mqtt.emqx1.conn_properties.receive_maximum=10
> bridges.mqtt.emqx1.conn_properties.session_expiry_interval=86400
> bridges.mqtt.emqx1.forwards.1.local_topic="test"
> bridges.mqtt.emqx1.forwards.1.remote_topic="7e18aa9a-ed43-11ee-a8e2-0662bd2efbf9/test"
> bridges.mqtt.emqx1.subscription.1.local_topic="test"
> bridges.mqtt.emqx1.subscription.1.remote_topic="7e18aa9a-ed43-11ee-a8e2-0662bd2efbf9/test"
> bridges.mqtt.emqx1.subscription.1.qos=2
> 
> bridges.mqtt.cache.disk_cache_size=102400
> bridges.mqtt.cache.mounted_file_path="/var/nanomq"
> bridges.mqtt.cache.flush_mem_threshold=1
> bridges.mqtt.cache.resend_interval=3000
> 
> http_server.port = 8081
> http_server.limit_conn = 2
> http_server.username = admin
> http_server.password = public
> http_server.auth_type = basic
> #http_server.jwt.public.keyfile = "/etc/certs/jwt/jwtRS256.key.pub"
> 
> log.to = [file, console]
> log.level = warn
> log.dir = "/var/log"
> log.file = "nanomq.log"
> log.rotation.size = 10MB
> log.rotation.count = 5
> 
> auth.allow_anonymous = true
> auth.no_match = allow
> auth.deny_action = ignore
> auth.cache.max_size = 32
> auth.cache.ttl = 1m

--- end body ---

None