1 /*
2  * Copyright (c) 2021 Huawei Device Co., Ltd.
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at
6  *
7  *     http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15 
16 #include "single_ver_sync_task_context.h"
17 
18 #include <algorithm>
19 #include "db_common.h"
20 #include "db_dfx_adapter.h"
21 #include "db_errno.h"
22 #include "isyncer.h"
23 #include "log_print.h"
24 #include "single_ver_sync_state_machine.h"
25 #include "single_ver_sync_target.h"
26 #include "sync_types.h"
27 
28 namespace DistributedDB {
SingleVerSyncTaskContext()29 SingleVerSyncTaskContext::SingleVerSyncTaskContext()
30     : SyncTaskContext(),
31       token_(nullptr),
32       endMark_(0),
33       needClearRemoteStaleData_(false)
34 {}
35 
~SingleVerSyncTaskContext()36 SingleVerSyncTaskContext::~SingleVerSyncTaskContext()
37 {
38     token_ = nullptr;
39     subManager_ = nullptr;
40 }
41 
Initialize(const std::string & deviceId,ISyncInterface * syncInterface,const std::shared_ptr<Metadata> & metadata,ICommunicator * communicator)42 int SingleVerSyncTaskContext::Initialize(const std::string &deviceId, ISyncInterface *syncInterface,
43     const std::shared_ptr<Metadata> &metadata, ICommunicator *communicator)
44 {
45     if (deviceId.empty() || syncInterface == nullptr || metadata == nullptr ||
46         communicator == nullptr) {
47         return -E_INVALID_ARGS;
48     }
49     stateMachine_ = new (std::nothrow) SingleVerSyncStateMachine;
50     if (stateMachine_ == nullptr) {
51         return -E_OUT_OF_MEMORY;
52     }
53     deviceId_ = deviceId;
54     std::vector<uint8_t> dbIdentifier = syncInterface->GetIdentifier();
55     dbIdentifier.resize(3); // only show 3 bytes
56     syncActionName_ = DBDfxAdapter::SYNC_ACTION + "_" +
57         DBCommon::VectorToHexString(dbIdentifier) + "_" + deviceId_.c_str();
58     TimerAction timeOutCallback;
59     int errCode = stateMachine_->Initialize(this, syncInterface, metadata, communicator);
60     if (errCode != E_OK) {
61         LOGE("[SingleVerSyncTaskContext] stateMachine Initialize failed, err %d.", errCode);
62         goto ERROR_OUT;
63     }
64 
65     timeHelper_ = std::make_unique<TimeHelper>();
66     errCode = timeHelper_->Initialize(syncInterface, metadata);
67     if (errCode != E_OK) {
68         LOGE("[SingleVerSyncTaskContext] timeHelper Initialize failed, err %d.", errCode);
69         goto ERROR_OUT;
70     }
71     timeOutCallback = [stateMachine = static_cast<SingleVerSyncStateMachine *>(stateMachine_)](TimerId timerId) {
72         return stateMachine->TimeoutCallback(timerId);
73     };
74     SetTimeoutCallback(timeOutCallback);
75 
76     syncInterface_ = syncInterface;
77     communicator_ = communicator;
78     taskExecStatus_ = INIT;
79     OnKill([this]() { this->KillWait(); });
80     {
81         std::lock_guard<std::mutex> lock(synTaskContextSetLock_);
82         synTaskContextSet_.insert(this);
83     }
84     return errCode;
85 
86 ERROR_OUT:
87     delete stateMachine_;
88     stateMachine_ = nullptr;
89     return errCode;
90 }
91 
AddSyncOperation(SyncOperation * operation)92 int SingleVerSyncTaskContext::AddSyncOperation(SyncOperation *operation)
93 {
94     if (operation == nullptr) {
95         return -E_INVALID_ARGS;
96     }
97 
98     // If auto sync, just update the end watermark
99     if (operation->IsAutoSync()) {
100         std::lock_guard<std::mutex> lock(targetQueueLock_);
101         bool isQuerySync = operation->IsQuerySync();
102         std::string queryId = operation->GetQueryId();
103         auto iter = std::find_if(requestTargetQueue_.begin(), requestTargetQueue_.end(),
104             [isQuerySync, queryId](const ISyncTarget *target) {
105             if (target == nullptr) {
106                 return false;
107             }
108             if (isQuerySync) {
109                 SyncOperation *tmpOperation = nullptr;
110                 target->GetSyncOperation(tmpOperation);
111                 return (tmpOperation != nullptr && tmpOperation->GetQueryId() == queryId) && target->IsAutoSync();
112             }
113             return target->IsAutoSync();
114         });
115         if (iter != requestTargetQueue_.end()) {
116             static_cast<SingleVerSyncTarget *>(*iter)->SetEndWaterMark(timeHelper_->GetTime());
117             operation->SetStatus(deviceId_, SyncOperation::OP_FINISHED_ALL);
118             return E_OK;
119         }
120     }
121 
122     auto *newTarget = new (std::nothrow) SingleVerSyncTarget;
123     if (newTarget == nullptr) {
124         return -E_OUT_OF_MEMORY;
125     }
126     newTarget->SetSyncOperation(operation);
127     Timestamp timstamp = timeHelper_->GetTime();
128     newTarget->SetEndWaterMark(timstamp);
129     newTarget->SetTaskType(ISyncTarget::REQUEST);
130     AddSyncTarget(newTarget);
131     return E_OK;
132 }
133 
SetEndMark(WaterMark endMark)134 void SingleVerSyncTaskContext::SetEndMark(WaterMark endMark)
135 {
136     endMark_ = endMark;
137 }
138 
GetEndMark() const139 WaterMark SingleVerSyncTaskContext::GetEndMark() const
140 {
141     return endMark_;
142 }
143 
GetContinueToken(ContinueToken & outToken) const144 void SingleVerSyncTaskContext::GetContinueToken(ContinueToken &outToken) const
145 {
146     outToken = token_;
147 }
148 
SetContinueToken(ContinueToken token)149 void SingleVerSyncTaskContext::SetContinueToken(ContinueToken token)
150 {
151     token_ = token;
152     return;
153 }
154 
ReleaseContinueToken()155 void SingleVerSyncTaskContext::ReleaseContinueToken()
156 {
157     if (token_ != nullptr) {
158         static_cast<SyncGenericInterface *>(syncInterface_)->ReleaseContinueToken(token_);
159         token_ = nullptr;
160     }
161 }
162 
PopResponseTarget(SingleVerSyncTarget & target)163 int SingleVerSyncTaskContext::PopResponseTarget(SingleVerSyncTarget &target)
164 {
165     std::lock_guard<std::mutex> lock(targetQueueLock_);
166     LOGD("[SingleVerSyncTaskContext] GetFrontExtWaterMark size = %zu", responseTargetQueue_.size());
167     if (!responseTargetQueue_.empty()) {
168         ISyncTarget *tmpTarget = responseTargetQueue_.front();
169         responseTargetQueue_.pop_front();
170         target = *(static_cast<SingleVerSyncTarget *>(tmpTarget));
171         delete tmpTarget;
172         tmpTarget = nullptr;
173         return E_OK;
174     }
175     return -E_LENGTH_ERROR;
176 }
177 
GetRspTargetQueueSize() const178 int SingleVerSyncTaskContext::GetRspTargetQueueSize() const
179 {
180     std::lock_guard<std::mutex> lock(targetQueueLock_);
181     return responseTargetQueue_.size();
182 }
183 
SetResponseSessionId(uint32_t responseSessionId)184 void SingleVerSyncTaskContext::SetResponseSessionId(uint32_t responseSessionId)
185 {
186     responseSessionId_ = responseSessionId;
187 }
188 
GetResponseSessionId() const189 uint32_t SingleVerSyncTaskContext::GetResponseSessionId() const
190 {
191     return responseSessionId_;
192 }
193 
CopyTargetData(const ISyncTarget * target,const TaskParam & taskParam)194 void SingleVerSyncTaskContext::CopyTargetData(const ISyncTarget *target, const TaskParam &taskParam)
195 {
196     const SingleVerSyncTarget *targetTmp = static_cast<const SingleVerSyncTarget *>(target);
197     SyncTaskContext::CopyTargetData(target, taskParam);
198     mode_ = targetTmp->GetMode();
199     endMark_ = targetTmp->GetEndWaterMark();
200     if (mode_ == SyncModeType::RESPONSE_PULL) {
201         responseSessionId_ = targetTmp->GetResponseSessionId();
202     }
203     {
204         std::lock_guard<std::mutex> autoLock(queryMutex_);
205         query_ = targetTmp->GetQuery();
206     }
207     isQuerySync_ = targetTmp->IsQuerySync();
208 }
209 
Clear()210 void SingleVerSyncTaskContext::Clear()
211 {
212     retryTime_ = 0;
213     ClearSyncOperation();
214     SyncTaskContext::Clear();
215     SetMode(SyncModeType::INVALID_MODE);
216     syncId_ = 0;
217     isAutoSync_ = false;
218     SetOperationStatus(SyncOperation::OP_WAITING);
219     SetEndMark(0);
220     SetResponseSessionId(0);
221     {
222         std::lock_guard<std::mutex> autoLock(queryMutex_);
223         query_ = QuerySyncObject();
224     }
225     isQuerySync_ = false;
226 }
227 
Abort(int status)228 void SingleVerSyncTaskContext::Abort(int status)
229 {
230     {
231         std::lock_guard<std::mutex> lock(operationLock_);
232         if (syncOperation_ != nullptr) {
233             syncOperation_->SetStatus(deviceId_, status, GetCommErrCode());
234             if ((status >= SyncOperation::OP_FINISHED_ALL)) {
235                 UnlockObj();
236                 if (syncOperation_->CheckIsAllFinished()) {
237                     syncOperation_->Finished();
238                 }
239                 LockObj();
240             }
241         }
242     }
243     StopFeedDogForSync(SyncDirectionFlag::SEND);
244     StopFeedDogForSync(SyncDirectionFlag::RECEIVE);
245     Clear();
246 }
247 
ClearAllSyncTask()248 void SingleVerSyncTaskContext::ClearAllSyncTask()
249 {
250     // clear request queue sync task and responsequeue first.
251     std::list<ISyncTarget *> targetQueue;
252     {
253         std::lock_guard<std::mutex> lock(targetQueueLock_);
254         LOGI("[SingleVerSyncTaskContext] request taskcount=%zu, responsecount=%zu", requestTargetQueue_.size(),
255             responseTargetQueue_.size());
256         while (!requestTargetQueue_.empty()) {
257             ISyncTarget *tmpTarget = requestTargetQueue_.front();
258             requestTargetQueue_.pop_front();
259             SyncOperation *tmpInfOperation = nullptr;
260             tmpTarget->GetSyncOperation(tmpInfOperation);
261             RefObject::IncObjRef(tmpInfOperation);
262             targetQueue.push_back(tmpTarget);
263         }
264         while (!responseTargetQueue_.empty()) {
265             ISyncTarget *tmpTarget = responseTargetQueue_.front();
266             responseTargetQueue_.pop_front();
267             delete tmpTarget;
268             tmpTarget = nullptr;
269         }
270     }
271     while (!targetQueue.empty()) {
272         ISyncTarget *target = targetQueue.front();
273         targetQueue.pop_front();
274         SyncOperation *tmpOperation = nullptr;
275         target->GetSyncOperation(tmpOperation);
276         if (tmpOperation == nullptr) {
277             LOGE("[ClearAllSyncTask] tmpOperation is nullptr");
278             continue; // not exit this scene
279         }
280         LOGI("[SingleVerSyncTaskContext] killing syncId=%d,dev=%s", tmpOperation->GetSyncId(), STR_MASK(deviceId_));
281         if (target->IsAutoSync()) {
282             tmpOperation->SetStatus(deviceId_, SyncOperation::OP_FINISHED_ALL);
283         } else {
284             tmpOperation->SetStatus(deviceId_, SyncOperation::OP_COMM_ABNORMAL);
285         }
286         if (tmpOperation->CheckIsAllFinished()) {
287             tmpOperation->Finished();
288         }
289         delete target;
290         target = nullptr;
291         RefObject::DecObjRef(tmpOperation);
292     }
293     if (GetTaskExecStatus() == SyncTaskContext::RUNNING) {
294         // clear syncing task.
295         stateMachine_->CommErrAbort();
296         SetCommFailErrCode(static_cast<int>(COMM_FAILURE));
297     }
298     // reset last push status for sync merge
299     ResetLastPushTaskStatus();
300 }
301 
EnableClearRemoteStaleData(bool enable)302 void SingleVerSyncTaskContext::EnableClearRemoteStaleData(bool enable)
303 {
304     needClearRemoteStaleData_ = enable;
305 }
306 
IsNeedClearRemoteStaleData() const307 bool SingleVerSyncTaskContext::IsNeedClearRemoteStaleData() const
308 {
309     return needClearRemoteStaleData_;
310 }
311 
StartFeedDogForSync(uint32_t time,SyncDirectionFlag flag)312 bool SingleVerSyncTaskContext::StartFeedDogForSync(uint32_t time, SyncDirectionFlag flag)
313 {
314     return stateMachine_->StartFeedDogForSync(time, flag);
315 }
316 
StopFeedDogForSync(SyncDirectionFlag flag)317 void SingleVerSyncTaskContext::StopFeedDogForSync(SyncDirectionFlag flag)
318 {
319     stateMachine_->StopFeedDogForSync(flag);
320 }
321 
IsReceiveWaterMarkErr() const322 bool SingleVerSyncTaskContext::IsReceiveWaterMarkErr() const
323 {
324     return isReceiveWaterMarkErr_;
325 }
326 
SetReceiveWaterMarkErr(bool isErr)327 void SingleVerSyncTaskContext::SetReceiveWaterMarkErr(bool isErr)
328 {
329     isReceiveWaterMarkErr_ = isErr;
330 }
331 
SetRemoteSeccurityOption(SecurityOption secOption)332 void SingleVerSyncTaskContext::SetRemoteSeccurityOption(SecurityOption secOption)
333 {
334     std::lock_guard<std::mutex> autoLock(securityOptionMutex_);
335     remoteSecOption_ = secOption;
336 }
337 
GetRemoteSeccurityOption() const338 SecurityOption SingleVerSyncTaskContext::GetRemoteSeccurityOption() const
339 {
340     std::lock_guard<std::mutex> autoLock(securityOptionMutex_);
341     return remoteSecOption_;
342 }
343 
SetReceivcPermitCheck(bool isChecked)344 void SingleVerSyncTaskContext::SetReceivcPermitCheck(bool isChecked)
345 {
346     isReceivcPermitChecked_ = isChecked;
347 }
348 
GetReceivcPermitCheck() const349 bool SingleVerSyncTaskContext::GetReceivcPermitCheck() const
350 {
351     return isReceivcPermitChecked_;
352 }
353 
SetSendPermitCheck(bool isChecked)354 void SingleVerSyncTaskContext::SetSendPermitCheck(bool isChecked)
355 {
356     isSendPermitChecked_ = isChecked;
357 }
358 
GetSendPermitCheck() const359 bool SingleVerSyncTaskContext::GetSendPermitCheck() const
360 {
361     return isSendPermitChecked_;
362 }
363 
IsSkipTimeoutError(int errCode) const364 bool SingleVerSyncTaskContext::IsSkipTimeoutError(int errCode) const
365 {
366     if (errCode == -E_TIMEOUT && IsSyncTaskNeedRetry() && (GetRetryTime() < GetSyncRetryTimes())) { // LCOV_EXCL_BR_LINE
367         LOGE("[SingleVerSyncTaskContext] send message timeout error occurred");
368         return true;
369     } else {
370         return false;
371     }
372 }
373 
FindResponseSyncTarget(uint32_t responseSessionId) const374 bool SingleVerSyncTaskContext::FindResponseSyncTarget(uint32_t responseSessionId) const
375 {
376     std::lock_guard<std::mutex> lock(targetQueueLock_);
377     auto iter = std::find_if(responseTargetQueue_.begin(), responseTargetQueue_.end(),
378         [responseSessionId](const ISyncTarget *target) {
379             return target->GetResponseSessionId() == responseSessionId;
380         });
381     if (iter == responseTargetQueue_.end()) {
382         return false;
383     }
384     return true;
385 }
386 
SetQuery(const QuerySyncObject & query)387 void SingleVerSyncTaskContext::SetQuery(const QuerySyncObject &query)
388 {
389     std::lock_guard<std::mutex> autoLock(queryMutex_);
390     query_ = query;
391 }
392 
GetQuery() const393 QuerySyncObject SingleVerSyncTaskContext::GetQuery() const
394 {
395     std::lock_guard<std::mutex> autoLock(queryMutex_);
396     return query_;
397 }
398 
SetQuerySync(bool isQuerySync)399 void SingleVerSyncTaskContext::SetQuerySync(bool isQuerySync)
400 {
401     isQuerySync_ = isQuerySync;
402 }
403 
IsQuerySync() const404 bool SingleVerSyncTaskContext::IsQuerySync() const
405 {
406     return isQuerySync_;
407 }
408 
GetRemoteCompressAlgo() const409 std::set<CompressAlgorithm> SingleVerSyncTaskContext::GetRemoteCompressAlgo() const
410 {
411     std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
412     std::set<CompressAlgorithm> compressAlgoSet;
413     for (const auto &algo : SyncConfig::COMPRESSALGOMAP) {
414         if (remoteDbAbility_.GetAbilityItem(algo.second) == SUPPORT_MARK) {
415             compressAlgoSet.insert(static_cast<CompressAlgorithm>(algo.first));
416         }
417     }
418     return compressAlgoSet;
419 }
420 
GetRemoteCompressAlgoStr() const421 std::string SingleVerSyncTaskContext::GetRemoteCompressAlgoStr() const
422 {
423     static std::map<CompressAlgorithm, std::string> algoMap = {{CompressAlgorithm::ZLIB, "zlib"}};
424     std::set<CompressAlgorithm> remoteCompressAlgoSet = GetRemoteCompressAlgo();
425     if (remoteCompressAlgoSet.empty()) {
426         return "none";
427     }
428     std::string currentAlgoStr;
429     for (const auto &algo : remoteCompressAlgoSet) {
430         auto iter = algoMap.find(algo);
431         if (iter != algoMap.end()) {
432             currentAlgoStr += algoMap[algo] + ",";
433         }
434     }
435     if (currentAlgoStr.empty()) {
436         return "";
437     }
438     return currentAlgoStr.substr(0, currentAlgoStr.length() - 1);
439 }
440 
SetDbAbility(DbAbility & remoteDbAbility)441 void SingleVerSyncTaskContext::SetDbAbility(DbAbility &remoteDbAbility)
442 {
443     {
444         std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
445         remoteDbAbility_ = remoteDbAbility;
446     }
447     LOGI("[SingleVerSyncTaskContext] set dev=%s compressAlgo=%s, IsSupAllPredicateQuery=%u,"
448         "IsSupSubscribeQuery=%u, inKeys=%u",
449         STR_MASK(GetDeviceId()), GetRemoteCompressAlgoStr().c_str(),
450         remoteDbAbility.GetAbilityItem(SyncConfig::ALLPREDICATEQUERY),
451         remoteDbAbility.GetAbilityItem(SyncConfig::SUBSCRIBEQUERY),
452         remoteDbAbility.GetAbilityItem(SyncConfig::INKEYS_QUERY));
453 }
454 
ChooseCompressAlgo() const455 CompressAlgorithm SingleVerSyncTaskContext::ChooseCompressAlgo() const
456 {
457     std::set<CompressAlgorithm> remoteAlgo = GetRemoteCompressAlgo();
458     if (remoteAlgo.empty()) {
459         return CompressAlgorithm::NONE;
460     }
461     std::set<CompressAlgorithm> localAlgorithmSet;
462     (void)(static_cast<SyncGenericInterface *>(syncInterface_))->GetCompressionAlgo(localAlgorithmSet);
463     std::set<CompressAlgorithm> algoIntersection;
464     set_intersection(remoteAlgo.begin(), remoteAlgo.end(), localAlgorithmSet.begin(), localAlgorithmSet.end(),
465         inserter(algoIntersection, algoIntersection.begin()));
466     if (algoIntersection.empty()) {
467         return CompressAlgorithm::NONE;
468     }
469     return *(algoIntersection.begin());
470 }
471 
IsNotSupportAbility(const AbilityItem & abilityItem) const472 bool SingleVerSyncTaskContext::IsNotSupportAbility(const AbilityItem &abilityItem) const
473 {
474     std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
475     return remoteDbAbility_.GetAbilityItem(abilityItem) != SUPPORT_MARK;
476 }
477 
SetSubscribeManager(std::shared_ptr<SubscribeManager> & subManager)478 void SingleVerSyncTaskContext::SetSubscribeManager(std::shared_ptr<SubscribeManager> &subManager)
479 {
480     subManager_ = subManager;
481 }
482 
GetSubscribeManager() const483 std::shared_ptr<SubscribeManager> SingleVerSyncTaskContext::GetSubscribeManager() const
484 {
485     return subManager_;
486 }
DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncTaskContext)487 DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncTaskContext)
488 
489 bool SingleVerSyncTaskContext::IsCurrentSyncTaskCanBeSkipped() const
490 {
491     SyncOperation *operation = GetAndIncSyncOperation();
492     bool res = IsCurrentSyncTaskCanBeSkippedInner(operation);
493     RefObject::DecObjRef(operation);
494     return res;
495 }
496 
SaveLastPushTaskExecStatus(int finalStatus)497 void SingleVerSyncTaskContext::SaveLastPushTaskExecStatus(int finalStatus)
498 {
499     if (IsTargetQueueEmpty()) {
500         LOGD("sync que is empty, reset last push status");
501         ResetLastPushTaskStatus();
502         return;
503     }
504     if (mode_ == SyncModeType::PUSH || mode_ == SyncModeType::PUSH_AND_PULL || mode_ == SyncModeType::RESPONSE_PULL) {
505         lastFullSyncTaskStatus_ = finalStatus;
506     } else if (mode_ == SyncModeType::QUERY_PUSH || mode_ == SyncModeType::QUERY_PUSH_PULL) {
507         std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
508         lastQuerySyncTaskStatusMap_[syncOperation_->GetQueryId()] = finalStatus;
509     }
510 }
511 
GetCorrectedSendWaterMarkForCurrentTask(const SyncOperation * operation,uint64_t & waterMark) const512 int SingleVerSyncTaskContext::GetCorrectedSendWaterMarkForCurrentTask(const SyncOperation *operation,
513     uint64_t &waterMark) const
514 {
515     if (operation != nullptr && operation->IsQuerySync()) {
516         LOGD("Is QuerySync");
517         int errCode = static_cast<SingleVerSyncStateMachine *>(stateMachine_)->GetSendQueryWaterMark(
518             operation->GetQueryId(), deviceId_,
519             lastFullSyncTaskStatus_ == SyncOperation::OP_FINISHED_ALL, waterMark);
520         if (errCode != E_OK) {
521             return errCode;
522         }
523     } else {
524         LOGD("Not QuerySync");
525         static_cast<SingleVerSyncStateMachine *>(stateMachine_)->GetLocalWaterMark(deviceId_, waterMark);
526     }
527     return E_OK;
528 }
529 
ResetLastPushTaskStatus()530 void SingleVerSyncTaskContext::ResetLastPushTaskStatus()
531 {
532     lastFullSyncTaskStatus_ = SyncOperation::OP_WAITING;
533     std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
534     lastQuerySyncTaskStatusMap_.clear();
535 }
536 
SetCommNormal(bool isCommNormal)537 void SingleVerSyncTaskContext::SetCommNormal(bool isCommNormal)
538 {
539     isCommNormal_ = isCommNormal;
540 }
541 
IsCurrentSyncTaskCanBeSkippedInner(const SyncOperation * operation) const542 bool SingleVerSyncTaskContext::IsCurrentSyncTaskCanBeSkippedInner(const SyncOperation *operation) const
543 {
544     if (mode_ == SyncModeType::PUSH) {
545         if (lastFullSyncTaskStatus_ != SyncOperation::OP_FINISHED_ALL) {
546             return false;
547         }
548         if (operation == nullptr) {
549             return true;
550         }
551     } else if (mode_ == SyncModeType::QUERY_PUSH) {
552         if (operation == nullptr) {
553             return true;
554         }
555         std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
556         auto it = lastQuerySyncTaskStatusMap_.find(operation->GetQueryId());
557         if (it == lastQuerySyncTaskStatusMap_.end()) {
558             // no last query_push and push
559             if (lastFullSyncTaskStatus_ != SyncOperation::OP_FINISHED_ALL) {
560                 LOGD("no prev query push or successful prev push");
561                 return false;
562             }
563         } else {
564             if (it->second != SyncOperation::OP_FINISHED_ALL) {
565                 LOGD("last query push status = %d.", it->second);
566                 return false;
567             }
568         }
569     } else {
570         return false;
571     }
572 
573     Timestamp maxTimestampInDb;
574     syncInterface_->GetMaxTimestamp(maxTimestampInDb);
575     uint64_t localWaterMark = 0;
576     int errCode = GetCorrectedSendWaterMarkForCurrentTask(operation, localWaterMark);
577     if (errCode != E_OK) {
578         LOGE("GetLocalWaterMark in state machine failed: %d", errCode);
579         return false;
580     }
581     if (localWaterMark > maxTimestampInDb) {
582         LOGI("skip current push task, deviceId_ = %s, localWaterMark = %" PRIu64 ", maxTimestampInDb = %" PRIu64,
583             STR_MASK(deviceId_), localWaterMark, maxTimestampInDb);
584         return true;
585     }
586     return false;
587 }
588 
StartFeedDogForGetData(uint32_t sessionId)589 void SingleVerSyncTaskContext::StartFeedDogForGetData(uint32_t sessionId)
590 {
591     stateMachine_->StartFeedDogForGetData(sessionId);
592 }
593 
StopFeedDogForGetData()594 void SingleVerSyncTaskContext::StopFeedDogForGetData()
595 {
596     stateMachine_->StopFeedDogForGetData();
597 }
598 
GetResponseTaskCount()599 int32_t SingleVerSyncTaskContext::GetResponseTaskCount()
600 {
601     std::lock_guard<std::mutex> autoLock(targetQueueLock_);
602     int32_t taskCount = static_cast<int32_t>(responseTargetQueue_.size());
603     if (responseSessionId_ != 0) {
604         taskCount++;
605     }
606     return taskCount;
607 }
608 } // namespace DistributedDB
609