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_state_machine.h"
17 
18 #include <cmath>
19 #include <climits>
20 #include <algorithm>
21 
22 #include "db_errno.h"
23 #include "log_print.h"
24 #include "sync_operation.h"
25 #include "message_transform.h"
26 #include "sync_types.h"
27 #include "db_common.h"
28 #include "runtime_context.h"
29 #include "performance_analysis.h"
30 #include "single_ver_sync_target.h"
31 #include "single_ver_data_sync.h"
32 #include "single_ver_data_sync_utils.h"
33 
34 namespace DistributedDB {
35 using Event = SingleVerSyncStateMachine::Event;
36 using State = SingleVerSyncStateMachine::State;
37 namespace {
38     // used for state switch table
39     const int CURRENT_STATE_INDEX = 0;
40     const int EVENT_INDEX = 1;
41     const int OUTPUT_STATE_INDEX = 2;
42 
43     // drop v1 and v2 table by one optimize, dataSend mode in all version go with slide window mode.
44     // State switch table v3, has three columns, CurrentState, Event, and OutSate
45     const std::vector<std::vector<uint8_t>> STATE_SWITCH_TABLE_V3 = {
46         {State::IDLE, Event::START_SYNC_EVENT, State::TIME_SYNC},
47 
48         // In TIME_SYNC state
49         {State::TIME_SYNC, Event::TIME_SYNC_FINISHED_EVENT, State::ABILITY_SYNC},
50         {State::TIME_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
51         {State::TIME_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
52 
53         // In ABILITY_SYNC state, compare version num and schema
54         {State::ABILITY_SYNC, Event::VERSION_NOT_SUPPOR_EVENT, State::INNER_ERR},
55         {State::ABILITY_SYNC, Event::ABILITY_SYNC_FINISHED_EVENT, State::START_INITIACTIVE_DATA_SYNC},
56         {State::ABILITY_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
57         {State::ABILITY_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
58         {State::ABILITY_SYNC, Event::CONTROL_CMD_EVENT, State::SYNC_CONTROL_CMD},
59 
60         // In START_INITIACTIVE_DATA_SYNC state, send a sync request, and send first packt of data sync
61         {State::START_INITIACTIVE_DATA_SYNC, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
62         {State::START_INITIACTIVE_DATA_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
63         {State::START_INITIACTIVE_DATA_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
64         {State::START_INITIACTIVE_DATA_SYNC, Event::SEND_FINISHED_EVENT, State::START_PASSIVE_DATA_SYNC},
65         {State::START_INITIACTIVE_DATA_SYNC, Event::RE_SEND_DATA_EVENT, State::START_INITIACTIVE_DATA_SYNC},
66         {State::START_INITIACTIVE_DATA_SYNC, Event::NEED_TIME_SYNC_EVENT, State::TIME_SYNC},
67 
68         // In START_PASSIVE_DATA_SYNC state, do response pull request, and send first packt of data sync
69         {State::START_PASSIVE_DATA_SYNC, Event::SEND_FINISHED_EVENT, State::START_PASSIVE_DATA_SYNC},
70         {State::START_PASSIVE_DATA_SYNC, Event::RESPONSE_TASK_FINISHED_EVENT, State::WAIT_FOR_RECEIVE_DATA_FINISH},
71         {State::START_PASSIVE_DATA_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
72         {State::START_PASSIVE_DATA_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
73         {State::START_PASSIVE_DATA_SYNC, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
74         {State::START_PASSIVE_DATA_SYNC, Event::RE_SEND_DATA_EVENT, State::START_PASSIVE_DATA_SYNC},
75         {State::START_PASSIVE_DATA_SYNC, Event::NEED_TIME_SYNC_EVENT, State::TIME_SYNC},
76 
77         // In WAIT_FOR_RECEIVE_DATA_FINISH,
78         {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::RECV_FINISHED_EVENT, State::SYNC_TASK_FINISHED},
79         {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::START_PULL_RESPONSE_EVENT, State::START_PASSIVE_DATA_SYNC},
80         {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
81         {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::INNER_ERR_EVENT, State::INNER_ERR},
82         {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
83 
84         {State::SYNC_CONTROL_CMD, Event::SEND_FINISHED_EVENT, State::SYNC_TASK_FINISHED},
85         {State::SYNC_CONTROL_CMD, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
86         {State::SYNC_CONTROL_CMD, Event::INNER_ERR_EVENT, State::INNER_ERR},
87         {State::SYNC_CONTROL_CMD, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
88 
89         // In SYNC_TASK_FINISHED,
90         {State::SYNC_TASK_FINISHED, Event::ALL_TASK_FINISHED_EVENT, State::IDLE},
91         {State::SYNC_TASK_FINISHED, Event::START_SYNC_EVENT, State::TIME_SYNC},
92 
93         // SYNC_TIME_OUT and INNE_ERR state, just do some exception resolve
94         {State::SYNC_TIME_OUT, Event::ANY_EVENT, State::SYNC_TASK_FINISHED},
95         {State::INNER_ERR, Event::ANY_EVENT, State::SYNC_TASK_FINISHED},
96     };
97 }
98 
99 std::mutex SingleVerSyncStateMachine::stateSwitchTableLock_;
100 std::vector<StateSwitchTable> SingleVerSyncStateMachine::stateSwitchTables_;
101 bool SingleVerSyncStateMachine::isStateSwitchTableInited_ = false;
102 
SingleVerSyncStateMachine()103 SingleVerSyncStateMachine::SingleVerSyncStateMachine()
104     : context_(nullptr),
105       syncInterface_(nullptr),
106       timeSync_(nullptr),
107       abilitySync_(nullptr),
108       dataSync_(nullptr),
109       currentRemoteVersionId_(0)
110 {
111 }
112 
~SingleVerSyncStateMachine()113 SingleVerSyncStateMachine::~SingleVerSyncStateMachine()
114 {
115     LOGD("~SingleVerSyncStateMachine");
116     Clear();
117 }
118 
Initialize(ISyncTaskContext * context,ISyncInterface * syncInterface,const std::shared_ptr<Metadata> & metaData,ICommunicator * communicator)119 int SingleVerSyncStateMachine::Initialize(ISyncTaskContext *context, ISyncInterface *syncInterface,
120     const std::shared_ptr<Metadata> &metaData, ICommunicator *communicator)
121 {
122     if ((context == nullptr) || (syncInterface == nullptr) || (metaData == nullptr) || (communicator == nullptr)) {
123         return -E_INVALID_ARGS;
124     }
125 
126     int errCode = SyncStateMachine::Initialize(context, syncInterface, metaData, communicator);
127     if (errCode != E_OK) {
128         return errCode;
129     }
130 
131     timeSync_ = std::make_shared<TimeSync>();
132     dataSync_ = std::make_shared<SingleVerDataSync>();
133     abilitySync_ = std::make_unique<AbilitySync>();
134     if ((timeSync_ == nullptr) || (dataSync_ == nullptr) || (abilitySync_ == nullptr)) {
135         timeSync_ = nullptr;
136         dataSync_ = nullptr;
137         abilitySync_ = nullptr;
138         return -E_OUT_OF_MEMORY;
139     }
140 
141     errCode = timeSync_->Initialize(communicator, metaData, syncInterface, context->GetDeviceId());
142     if (errCode != E_OK) {
143         goto ERROR_OUT;
144     }
145     errCode = dataSync_->Initialize(syncInterface, communicator, metaData, context->GetDeviceId());
146     if (errCode != E_OK) {
147         goto ERROR_OUT;
148     }
149     errCode = abilitySync_->Initialize(communicator, syncInterface, metaData, context->GetDeviceId());
150     if (errCode != E_OK) {
151         goto ERROR_OUT;
152     }
153     abilitySync_->InitAbilitySyncFinishStatus(*context);
154 
155     currentState_ = IDLE;
156     context_ = static_cast<SingleVerSyncTaskContext *>(context);
157     syncInterface_ = static_cast<SyncGenericInterface *>(syncInterface);
158 
159     InitStateSwitchTables();
160     InitStateMapping();
161     return E_OK;
162 
163 ERROR_OUT:
164     Clear();
165     return errCode;
166 }
167 
SyncStep()168 void SingleVerSyncStateMachine::SyncStep()
169 {
170     RefObject::IncObjRef(context_);
171     RefObject::IncObjRef(communicator_);
172     int errCode = RuntimeContext::GetInstance()->ScheduleTask([this] { SyncStepInnerLocked(); });
173     if (errCode != E_OK) {
174         LOGE("[StateMachine][SyncStep] Schedule SyncStep failed");
175         RefObject::DecObjRef(communicator_);
176         RefObject::DecObjRef(context_);
177     }
178 }
179 
ReceiveMessageCallback(Message * inMsg)180 int SingleVerSyncStateMachine::ReceiveMessageCallback(Message *inMsg)
181 {
182     int errCode = MessageCallbackPre(inMsg);
183     if (errCode != E_OK) {
184         LOGE("[StateMachine] message pre check failed");
185         return errCode;
186     }
187     switch (inMsg->GetMessageId()) {
188         case TIME_SYNC_MESSAGE:
189             errCode = TimeMarkSyncRecv(inMsg);
190             break;
191         case ABILITY_SYNC_MESSAGE:
192             errCode = AbilitySyncRecv(inMsg);
193             break;
194         case DATA_SYNC_MESSAGE:
195         case QUERY_SYNC_MESSAGE:
196             errCode = DataPktRecv(inMsg);
197             break;
198         case CONTROL_SYNC_MESSAGE:
199             errCode = ControlPktRecv(inMsg);
200             break;
201         default:
202             errCode = -E_NOT_SUPPORT;
203     }
204     return errCode;
205 }
206 
SyncStepInnerLocked()207 void SingleVerSyncStateMachine::SyncStepInnerLocked()
208 {
209     if (context_->IncUsedCount() != E_OK) {
210         goto SYNC_STEP_OUT;
211     }
212     {
213         std::lock_guard<std::mutex> lock(stateMachineLock_);
214         SyncStepInner();
215     }
216     context_->SafeExit();
217 
218 SYNC_STEP_OUT:
219     RefObject::DecObjRef(communicator_);
220     RefObject::DecObjRef(context_);
221 }
222 
SyncStepInner()223 void SingleVerSyncStateMachine::SyncStepInner()
224 {
225     Event event = INNER_ERR_EVENT;
226     do {
227         auto iter = stateMapping_.find(currentState_);
228         if (iter != stateMapping_.end()) {
229             event = static_cast<Event>(iter->second());
230         } else {
231             LOGE("[StateMachine][SyncStepInner] can not find state=%d,label=%s,dev=%s", currentState_,
232                 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
233             break;
234         }
235     } while (event != Event::WAIT_ACK_EVENT && SwitchMachineState(event) == E_OK && currentState_ != IDLE);
236 }
237 
SetCurStateErrStatus()238 void SingleVerSyncStateMachine::SetCurStateErrStatus()
239 {
240     currentState_ = State::INNER_ERR;
241 }
242 
StartSyncInner()243 int SingleVerSyncStateMachine::StartSyncInner()
244 {
245     PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
246     if (performance != nullptr) {
247         performance->StepTimeRecordStart(PT_TEST_RECORDS::RECORD_MACHINE_START_TO_PUSH_SEND);
248     }
249     int errCode = PrepareNextSyncTask();
250     if (errCode == E_OK) {
251         SwitchStateAndStep(Event::START_SYNC_EVENT);
252     }
253     return errCode;
254 }
255 
AbortInner()256 void SingleVerSyncStateMachine::AbortInner()
257 {
258     LOGE("[StateMachine][AbortInner] error occurred,abort,label=%s,dev=%s", dataSync_->GetLabel().c_str(),
259         STR_MASK(context_->GetDeviceId()));
260     if (context_->IsKilled()) {
261         dataSync_->ClearDataMsg();
262     }
263     dataSync_->ClearSyncStatus();
264     ContinueToken token;
265     context_->GetContinueToken(token);
266     if (token != nullptr) {
267         syncInterface_->ReleaseContinueToken(token);
268     }
269     context_->SetContinueToken(nullptr);
270     context_->Clear();
271 }
272 
GetStateSwitchTables() const273 const std::vector<StateSwitchTable> &SingleVerSyncStateMachine::GetStateSwitchTables() const
274 {
275     return stateSwitchTables_;
276 }
277 
PrepareNextSyncTask()278 int SingleVerSyncStateMachine::PrepareNextSyncTask()
279 {
280     int errCode = StartWatchDog();
281     if (errCode != E_OK && errCode != -E_UNEXPECTED_DATA) {
282         LOGE("[StateMachine][PrepareNextSyncTask] WatchDog start failed,err=%d", errCode);
283         return errCode;
284     }
285     if (errCode == -E_UNEXPECTED_DATA) {
286         LOGI("[PrepareNextSyncTask] timer already exists, reset the timer.");
287         (void)ResetWatchDog();
288     }
289 
290     if (currentState_ != State::IDLE && currentState_ != State::SYNC_TASK_FINISHED) {
291         LOGW("[StateMachine][PrepareNextSyncTask] PreSync may get an err, state=%" PRIu8 ",dev=%s",
292             currentState_, STR_MASK(context_->GetDeviceId()));
293         currentState_ = State::IDLE;
294     }
295     return E_OK;
296 }
297 
SendNotifyPacket(uint32_t sessionId,uint32_t sequenceId,uint32_t inMsgId)298 void SingleVerSyncStateMachine::SendNotifyPacket(uint32_t sessionId, uint32_t sequenceId, uint32_t inMsgId)
299 {
300     dataSync_->SendSaveDataNotifyPacket(context_,
301         std::min(context_->GetRemoteSoftwareVersion(), SOFTWARE_VERSION_CURRENT), sessionId, sequenceId, inMsgId);
302 }
303 
CommErrAbort(uint32_t sessionId)304 void SingleVerSyncStateMachine::CommErrAbort(uint32_t sessionId)
305 {
306     std::lock_guard<std::mutex> lock(stateMachineLock_);
307     uint32_t requestSessionId = context_->GetRequestSessionId();
308     if ((sessionId != 0) && ((requestSessionId == 0) || (sessionId != requestSessionId))) {
309         return;
310     }
311     context_->SetCommNormal(false);
312     if (SwitchMachineState(Event::INNER_ERR_EVENT) == E_OK) {
313         SyncStep();
314     }
315 }
316 
InitStateSwitchTables()317 void SingleVerSyncStateMachine::InitStateSwitchTables()
318 {
319     if (isStateSwitchTableInited_) {
320         return;
321     }
322 
323     std::lock_guard<std::mutex> lock(stateSwitchTableLock_);
324     if (isStateSwitchTableInited_) {
325         return;
326     }
327 
328     InitStateSwitchTable(SINGLE_VER_SYNC_PROCTOL_V3, STATE_SWITCH_TABLE_V3);
329     std::sort(stateSwitchTables_.begin(), stateSwitchTables_.end(),
330         [](const auto &tableA, const auto &tableB) {
331             return tableA.version > tableB.version;
332         }); // descending
333     isStateSwitchTableInited_ = true;
334 }
335 
InitStateSwitchTable(uint32_t version,const std::vector<std::vector<uint8_t>> & switchTable)336 void SingleVerSyncStateMachine::InitStateSwitchTable(uint32_t version,
337     const std::vector<std::vector<uint8_t>> &switchTable)
338 {
339     StateSwitchTable table;
340     table.version = version;
341     for (const auto &stateSwitch : switchTable) {
342         if (stateSwitch.size() <= OUTPUT_STATE_INDEX) {
343             LOGE("[StateMachine][InitSwitchTable] stateSwitch size err,size=%zu", stateSwitch.size());
344             return;
345         }
346         if (table.switchTable.count(stateSwitch[CURRENT_STATE_INDEX]) == 0) {
347             EventToState eventToState; // new EventToState
348             eventToState[stateSwitch[EVENT_INDEX]] = stateSwitch[OUTPUT_STATE_INDEX];
349             table.switchTable[stateSwitch[CURRENT_STATE_INDEX]] = eventToState;
350         } else { // key stateSwitch[CURRENT_STATE_INDEX] already has EventToState
351             EventToState &eventToState = table.switchTable[stateSwitch[CURRENT_STATE_INDEX]];
352             eventToState[stateSwitch[EVENT_INDEX]] = stateSwitch[OUTPUT_STATE_INDEX];
353         }
354     }
355     stateSwitchTables_.push_back(table);
356 }
357 
InitStateMapping()358 void SingleVerSyncStateMachine::InitStateMapping()
359 {
360     stateMapping_[TIME_SYNC] = [this] { return DoTimeSync(); };
361     stateMapping_[ABILITY_SYNC] = [this] { return DoAbilitySync(); };
362     stateMapping_[WAIT_FOR_RECEIVE_DATA_FINISH] = [this] { return DoWaitForDataRecv(); };
363     stateMapping_[SYNC_TASK_FINISHED] = [this] { return DoSyncTaskFinished(); };
364     stateMapping_[SYNC_TIME_OUT] = [this] { return DoTimeout(); };
365     stateMapping_[INNER_ERR] = [this] { return DoInnerErr(); };
366     stateMapping_[START_INITIACTIVE_DATA_SYNC] = [this] { return DoInitiactiveDataSyncWithSlidingWindow(); };
367     stateMapping_[START_PASSIVE_DATA_SYNC] = [this] { return DoPassiveDataSyncWithSlidingWindow(); };
368     stateMapping_[SYNC_CONTROL_CMD] = [this] { return DoInitiactiveControlSync(); };
369 }
370 
DoInitiactiveDataSyncWithSlidingWindow() const371 Event SingleVerSyncStateMachine::DoInitiactiveDataSyncWithSlidingWindow() const
372 {
373     LOGD("[StateMachine][activeDataSync] mode=%d,label=%s,dev=%s", context_->GetMode(),
374         dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
375     int errCode = E_OK;
376     switch (context_->GetMode()) {
377         case SyncModeType::PUSH:
378         case SyncModeType::QUERY_PUSH:
379             context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
380             errCode = dataSync_->SyncStart(context_->GetMode(), context_);
381             break;
382         case SyncModeType::PULL:
383         case SyncModeType::QUERY_PULL:
384             context_->SetOperationStatus(SyncOperation::OP_SEND_FINISHED);
385             errCode = dataSync_->SyncStart(context_->GetMode(), context_);
386             break;
387         case SyncModeType::PUSH_AND_PULL:
388         case SyncModeType::QUERY_PUSH_PULL:
389             errCode = dataSync_->SyncStart(context_->GetMode(), context_);
390             break;
391         case SyncModeType::RESPONSE_PULL:
392             errCode = dataSync_->SyncStart(context_->GetMode(), context_);
393             break;
394         default:
395             errCode = -E_NOT_SUPPORT;
396             break;
397     }
398     if (errCode == E_OK) {
399         return Event::WAIT_ACK_EVENT;
400     }
401     // once E_EKEYREVOKED error occurred, PUSH_AND_PULL mode should wait for ack to pull remote data.
402     if (SyncOperation::TransferSyncMode(context_->GetMode()) == SyncModeType::PUSH_AND_PULL &&
403         errCode == -E_EKEYREVOKED) {
404         return Event::WAIT_ACK_EVENT;
405     }
406 
407     // when response task step dataSync again while request task is running,  ignore the errCode
408     bool ignoreInnerErr = (context_->GetResponseSessionId() != 0 && context_->GetRequestSessionId() != 0);
409     Event event = TransformErrCodeToEvent(errCode);
410     return (ignoreInnerErr && event == INNER_ERR_EVENT) ? SEND_FINISHED_EVENT : event;
411 }
412 
DoPassiveDataSyncWithSlidingWindow()413 Event SingleVerSyncStateMachine::DoPassiveDataSyncWithSlidingWindow()
414 {
415     {
416         RefObject::AutoLock lock(context_);
417         if (context_->GetRspTargetQueueSize() != 0) {
418             PreStartPullResponse();
419         } else if (context_->GetResponseSessionId() == 0 ||
420             context_->GetRetryStatus() == static_cast<int>(SyncTaskContext::NO_NEED_RETRY)) {
421             return RESPONSE_TASK_FINISHED_EVENT;
422         }
423     }
424     int errCode = dataSync_->SyncStart(SyncModeType::RESPONSE_PULL, context_);
425     if (errCode != E_OK) {
426         LOGE("[SingleVerSyncStateMachine][DoPassiveDataSyncWithSlidingWindow] response pull send failed[%d]", errCode);
427         return RESPONSE_TASK_FINISHED_EVENT;
428     }
429     return Event::WAIT_ACK_EVENT;
430 }
431 
DoInitiactiveControlSync() const432 Event SingleVerSyncStateMachine::DoInitiactiveControlSync() const
433 {
434     LOGD("[StateMachine][activeControlSync] mode=%d,label=%s,dev=%s", context_->GetMode(),
435         dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
436     context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
437     int errCode = dataSync_->ControlCmdStart(context_);
438     if (errCode == E_OK) {
439         return Event::WAIT_ACK_EVENT;
440     }
441     context_->SetTaskErrCode(errCode);
442     return TransformErrCodeToEvent(errCode);
443 }
444 
HandleControlAckRecv(const Message * inMsg)445 int SingleVerSyncStateMachine::HandleControlAckRecv(const Message *inMsg)
446 {
447     std::lock_guard<std::mutex> lock(stateMachineLock_);
448     if (IsNeedResetWatchdog(inMsg)) {
449         (void)ResetWatchDog();
450     }
451     int errCode = dataSync_->ControlCmdAckRecv(context_, inMsg);
452     ControlAckRecvErrCodeHandle(errCode);
453     SwitchStateAndStep(TransformErrCodeToEvent(errCode));
454     return E_OK;
455 }
456 
DoWaitForDataRecv() const457 Event SingleVerSyncStateMachine::DoWaitForDataRecv() const
458 {
459     if (context_->GetRspTargetQueueSize() != 0) {
460         return START_PULL_RESPONSE_EVENT;
461     }
462     if (context_->GetOperationStatus() == SyncOperation::OP_FINISHED_ALL) {
463         return RECV_FINISHED_EVENT;
464     }
465     if (SyncOperation::TransferSyncMode(context_->GetMode()) == SyncModeType::PUSH_AND_PULL &&
466         context_->GetOperationStatus() == SyncOperation::OP_EKEYREVOKED_FAILURE &&
467         context_->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_2_0) {
468         return RECV_FINISHED_EVENT;
469     }
470     return Event::WAIT_ACK_EVENT;
471 }
472 
DoTimeSync() const473 Event SingleVerSyncStateMachine::DoTimeSync() const
474 {
475     timeSync_->SetTimeSyncFinishIfNeed();
476     if (timeSync_->IsNeedSync()) {
477         CommErrHandler handler = nullptr;
478         // Auto sync need do retry don't use errHandler to return.
479         if (!context_->IsAutoSync()) {
480             handler = [this, context = context_,
481                 requestSessionId = context_->GetRequestSessionId()](int ret, bool isDirectEnd) {
482                 SyncTaskContext::CommErrHandlerFunc(ret, context, requestSessionId, isDirectEnd);
483             };
484         }
485         int errCode = timeSync_->SyncStart(handler, context_->GetRequestSessionId());
486         if (errCode == E_OK) {
487             return Event::WAIT_ACK_EVENT;
488         }
489         context_->SetTaskErrCode(errCode);
490         return TransformErrCodeToEvent(errCode);
491     }
492 
493     return Event::TIME_SYNC_FINISHED_EVENT;
494 }
495 
DoAbilitySync() const496 Event SingleVerSyncStateMachine::DoAbilitySync() const
497 {
498     uint16_t remoteCommunicatorVersion = 0;
499     int errCode = communicator_->GetRemoteCommunicatorVersion(context_->GetDeviceId(), remoteCommunicatorVersion);
500     if (errCode != E_OK && errCode != -E_NOT_FOUND) {
501         LOGE("[StateMachine][DoAbilitySync] Get RemoteCommunicatorVersion errCode=%d", errCode);
502         return Event::INNER_ERR_EVENT;
503     }
504     // Fistr version, not support AbilitySync
505     if (remoteCommunicatorVersion == 0 && errCode != -E_NOT_FOUND) {
506         context_->SetRemoteSoftwareVersion(SOFTWARE_VERSION_EARLIEST);
507         return GetEventAfterTimeSync(context_->GetMode());
508     }
509     if (abilitySync_->GetAbilitySyncFinishedStatus()) {
510         return GetEventAfterTimeSync(context_->GetMode());
511     }
512 
513     CommErrHandler handler = [this, context = context_,
514         requestSessionId = context_->GetRequestSessionId()](int ret, bool isDirectEnd) {
515         SyncTaskContext::CommErrHandlerFunc(ret, context, requestSessionId, isDirectEnd);
516     };
517     LOGI("[StateMachine][AbilitySync] start abilitySync,label=%s,dev=%s", dataSync_->GetLabel().c_str(),
518         STR_MASK(context_->GetDeviceId()));
519     errCode = abilitySync_->SyncStart(context_->GetRequestSessionId(), context_->GetSequenceId(),
520         remoteCommunicatorVersion, handler, context_);
521     if (errCode != E_OK) {
522         LOGE("[StateMachine][DoAbilitySync] ability sync start failed,errCode=%d", errCode);
523         return TransformErrCodeToEvent(errCode);
524     }
525     return Event::WAIT_ACK_EVENT;
526 }
527 
GetEventAfterTimeSync(int mode) const528 Event SingleVerSyncStateMachine::GetEventAfterTimeSync(int mode) const
529 {
530     if (mode == SyncModeType::SUBSCRIBE_QUERY || mode == SyncModeType::UNSUBSCRIBE_QUERY) {
531         return Event::CONTROL_CMD_EVENT;
532     }
533     return Event::ABILITY_SYNC_FINISHED_EVENT;
534 }
535 
DoSyncTaskFinished()536 Event SingleVerSyncStateMachine::DoSyncTaskFinished()
537 {
538     StopWatchDog();
539     dataSync_->ClearSyncStatus();
540     RefObject::AutoLock lock(syncContext_);
541     int errCode = ExecNextTask();
542     if (errCode == E_OK) {
543         return Event::START_SYNC_EVENT;
544     }
545     return TransformErrCodeToEvent(errCode);
546 }
547 
DoTimeout()548 Event SingleVerSyncStateMachine::DoTimeout()
549 {
550     RefObject::AutoLock lock(context_);
551     if (context_->GetMode() == SyncModeType::SUBSCRIBE_QUERY) {
552         std::shared_ptr<SubscribeManager> subManager = context_->GetSubscribeManager();
553         if (subManager != nullptr) {
554             subManager->DeleteLocalSubscribeQuery(context_->GetDeviceId(), context_->GetQuery());
555         }
556     }
557     context_->Abort(SyncOperation::OP_TIMEOUT);
558     context_->Clear();
559     AbortInner();
560     return Event::ANY_EVENT;
561 }
562 
DoInnerErr()563 Event SingleVerSyncStateMachine::DoInnerErr()
564 {
565     RefObject::AutoLock lock(context_);
566     if (!context_->IsCommNormal()) {
567         if (context_->GetMode() == SyncModeType::SUBSCRIBE_QUERY) {
568             std::shared_ptr<SubscribeManager> subManager = context_->GetSubscribeManager();
569             if (subManager != nullptr) {
570                 subManager->DeleteLocalSubscribeQuery(context_->GetDeviceId(), context_->GetQuery());
571             }
572         }
573         context_->Abort(SyncOperation::OP_COMM_ABNORMAL);
574     } else {
575         int status = GetSyncOperationStatus(context_->GetTaskErrCode());
576         context_->Abort(status);
577     }
578     context_->Clear();
579     AbortInner();
580     return Event::ANY_EVENT;
581 }
582 
AbilitySyncRecv(const Message * inMsg)583 int SingleVerSyncStateMachine::AbilitySyncRecv(const Message *inMsg)
584 {
585     if (inMsg->GetMessageType() == TYPE_REQUEST) {
586         int errCode = abilitySync_->RequestRecv(inMsg, context_);
587         if (errCode != E_OK && inMsg->GetSessionId() == context_->GetResponseSessionId()) {
588             context_->SetTaskErrCode(errCode);
589             std::lock_guard<std::mutex> lock(stateMachineLock_);
590             SwitchStateAndStep(Event::INNER_ERR_EVENT);
591         }
592         return E_OK;
593     }
594     return AbilitySyncResponseRecv(inMsg);
595 }
596 
AbilitySyncResponseRecv(const Message * inMsg)597 int SingleVerSyncStateMachine::AbilitySyncResponseRecv(const Message *inMsg)
598 {
599     if (inMsg->GetMessageType() == TYPE_RESPONSE && AbilityMsgSessionIdCheck(inMsg)) {
600         std::lock_guard<std::mutex> lock(stateMachineLock_);
601         int errCode = abilitySync_->AckRecv(inMsg, context_);
602         (void)ResetWatchDog();
603         if (errCode == -E_ABILITY_SYNC_FINISHED) {
604             LOGI("[StateMachine][AbilitySyncRecv] ability sync finished with both kv,label=%s,dev=%s",
605                 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
606             abilitySync_->SetAbilitySyncFinishedStatus(true, *context_);
607             JumpStatusAfterAbilitySync(context_->GetMode());
608         } else if (errCode != E_OK) {
609             LOGE("[StateMachine][AbilitySyncRecv] handle ackRecv failed,errCode=%d", errCode);
610             SwitchStateAndStep(TransformErrCodeToEvent(errCode));
611         } else if (context_->GetRemoteSoftwareVersion() <= SOFTWARE_VERSION_RELEASE_2_0) {
612             abilitySync_->SetAbilitySyncFinishedStatus(true, *context_);
613             LOGI("[StateMachine][AbilitySyncRecv] ability Sync Finished,label=%s,dev=%s",
614                 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
615             currentRemoteVersionId_ = context_->GetRemoteSoftwareVersionId();
616             JumpStatusAfterAbilitySync(context_->GetMode());
617         }
618         return E_OK;
619     }
620     if (inMsg->GetMessageType() == TYPE_NOTIFY) {
621         return AbilitySyncNotifyRecv(inMsg);
622     }
623     LOGE("[StateMachine][AbilitySyncRecv] msg type invalid");
624     return -E_NOT_SUPPORT;
625 }
626 
HandleDataRequestRecv(const Message * inMsg)627 int SingleVerSyncStateMachine::HandleDataRequestRecv(const Message *inMsg)
628 {
629     TimeOffset offset = 0;
630     auto [systemOffset, senderLocalOffset] = SingleVerDataSyncUtils::GetTimeOffsetFromRequestMsg(inMsg);
631     int errCode = timeSync_->GenerateTimeOffsetIfNeed(systemOffset, senderLocalOffset);
632     if (errCode != E_OK) {
633         (void)dataSync_->SendDataAck(context_, inMsg, errCode, 0);
634         return errCode;
635     }
636     uint32_t timeout = communicator_->GetTimeout(context_->GetDeviceId());
637     // If message is data sync request, we should check timeoffset.
638     errCode = timeSync_->GetTimeOffset(offset, timeout);
639     if (errCode != E_OK) {
640         LOGE("[StateMachine][HandleDataRequestRecv] GetTimeOffset err! errCode=%d", errCode);
641         return errCode;
642     }
643     context_->SetTimeOffset(offset);
644     PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
645     if (performance != nullptr) {
646         performance->StepTimeRecordStart(PT_TEST_RECORDS::RECORD_DATA_REQUEST_RECV_TO_SEND_ACK);
647     }
648     DecRefCountOfFeedDogTimer(SyncDirectionFlag::RECEIVE);
649 
650     // RequestRecv will save data, it may cost a long time.
651     // So we need to send save data notify to keep remote alive.
652     bool isNeedStop = StartSaveDataNotify(inMsg->GetSessionId(), inMsg->GetSequenceId(), inMsg->GetMessageId());
653     {
654         std::lock_guard<std::mutex> lockWatchDog(stateMachineLock_);
655         if (IsNeedResetWatchdog(inMsg)) {
656             (void)ResetWatchDog();
657         }
658     }
659     WaterMark pullEndWaterkark = 0;
660     errCode = dataSync_->DataRequestRecv(context_, inMsg, pullEndWaterkark);
661     if (performance != nullptr) {
662         performance->StepTimeRecordEnd(PT_TEST_RECORDS::RECORD_DATA_REQUEST_RECV_TO_SEND_ACK);
663     }
664     // only higher than 102 version receive this errCode here.
665     // while both RequestSessionId is not equal,but get this errCode;slwr would seem to handle first secquencid.
666     // so while receive the same secquencid after abiitysync it wouldn't handle.
667     if (errCode == -E_NEED_ABILITY_SYNC) {
668         if (isNeedStop) {
669             StopSaveDataNotify();
670         }
671         return errCode;
672     }
673     {
674         std::lock_guard<std::mutex> lock(stateMachineLock_);
675         DataRecvErrCodeHandle(inMsg->GetSessionId(), errCode);
676         if (pullEndWaterkark > 0) {
677             AddPullResponseTarget(inMsg, pullEndWaterkark);
678         }
679     }
680     if (isNeedStop) {
681         StopSaveDataNotify();
682     }
683     return E_OK;
684 }
685 
HandleDataAckRecvWithSlidingWindow(int errCode,const Message * inMsg,bool ignoreInnerErr)686 void SingleVerSyncStateMachine::HandleDataAckRecvWithSlidingWindow(int errCode, const Message *inMsg,
687     bool ignoreInnerErr)
688 {
689     if (errCode == -E_RE_SEND_DATA) { // LOCAL_WATER_MARK_NOT_INIT
690         dataSync_->ClearSyncStatus();
691     }
692     if (errCode == -E_NO_DATA_SEND || errCode == -E_SEND_DATA) {
693         int ret = dataSync_->TryContinueSync(context_, inMsg);
694         if (ret == -E_FINISHED) {
695             SwitchStateAndStep(Event::SEND_FINISHED_EVENT);
696             return;
697         } else if (ret == E_OK) { // do nothing and waiting for all ack receive
698             return;
699         }
700         errCode = ret;
701     }
702     ResponsePullError(errCode, ignoreInnerErr);
703 }
704 
NeedAbilitySyncHandle()705 void SingleVerSyncStateMachine::NeedAbilitySyncHandle()
706 {
707     // if the remote device version num is overdue,
708     // mean the version num has been reset when syncing data,
709     // there should not clear the new version cache again.
710     if (currentRemoteVersionId_ == context_->GetRemoteSoftwareVersionId()) {
711         LOGI("[StateMachine] set remote version 0, currentRemoteVersionId_ = %" PRIu64, currentRemoteVersionId_);
712         context_->SetRemoteSoftwareVersion(0);
713     } else {
714         currentRemoteVersionId_ = context_->GetRemoteSoftwareVersionId();
715     }
716     abilitySync_->SetAbilitySyncFinishedStatus(false, *context_);
717 }
718 
HandleDataAckRecv(const Message * inMsg)719 int SingleVerSyncStateMachine::HandleDataAckRecv(const Message *inMsg)
720 {
721     if (inMsg->GetMessageType() == TYPE_RESPONSE) {
722         DecRefCountOfFeedDogTimer(SyncDirectionFlag::SEND);
723     }
724     std::lock_guard<std::mutex> lock(stateMachineLock_);
725     // Unfortunately we use stateMachineLock_ in many sync process
726     // So a bad ack will check before the lock and wait
727     // And then another process is running, it will get the lock.After this process, the ack became invalid.
728     // If we don't check ack again, it will be delivered to dataSyncer.
729     if (!IsPacketValid(inMsg)) {
730         return -E_INVALID_ARGS;
731     }
732     if (IsNeedResetWatchdog(inMsg)) {
733         (void)ResetWatchDog();
734     }
735     if (context_->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_2_0 && !dataSync_->AckPacketIdCheck(inMsg)) {
736         // packetId not match but sequence id matched scene, means resend map has be rebuilt
737         // this is old ack, should be dropped and wait for the same packetId sequence.
738         return E_OK;
739     }
740     // AckRecv will save meta data, it may cost a long time. if another thread is saving data
741     // So we need to send save data notify to keep remote alive.
742     // e.g. remote do pull sync
743     bool isNeedStop = StartSaveDataNotify(inMsg->GetSessionId(), inMsg->GetSequenceId(), inMsg->GetMessageId());
744     int errCode = dataSync_->AckRecv(context_, inMsg);
745     if (isNeedStop) {
746         StopSaveDataNotify();
747     }
748     if (errCode == -E_NEED_ABILITY_SYNC || errCode == -E_RE_SEND_DATA || errCode == -E_NEED_TIME_SYNC) {
749         StopFeedDogForSync(SyncDirectionFlag::SEND);
750     } else if (errCode == -E_SAVE_DATA_NOTIFY) {
751         return errCode;
752     }
753     // when this msg is from response task while request task is running,  ignore the errCode
754     bool ignoreInnerErr = inMsg->GetSessionId() == context_->GetResponseSessionId() &&
755         context_->GetRequestSessionId() != 0;
756     DataAckRecvErrCodeHandle(errCode, !ignoreInnerErr);
757     HandleDataAckRecvWithSlidingWindow(errCode, inMsg, ignoreInnerErr);
758     return errCode;
759 }
760 
DataPktRecv(Message * inMsg)761 int SingleVerSyncStateMachine::DataPktRecv(Message *inMsg)
762 {
763     PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
764     int errCode = E_OK;
765     switch (inMsg->GetMessageType()) {
766         case TYPE_REQUEST:
767             ScheduleMsgAndHandle(inMsg);
768             errCode = -E_NOT_NEED_DELETE_MSG;
769             break;
770         case TYPE_RESPONSE:
771         case TYPE_NOTIFY:
772             if (performance != nullptr) {
773                 performance->StepTimeRecordEnd(PT_TEST_RECORDS::RECORD_DATA_SEND_REQUEST_TO_ACK_RECV);
774                 performance->StepTimeRecordStart(PT_TEST_RECORDS::RECORD_ACK_RECV_TO_USER_CALL_BACK);
775             }
776             errCode = HandleDataAckRecv(inMsg);
777             break;
778         default:
779             errCode = -E_INVALID_ARGS;
780             break;
781     }
782     return errCode;
783 }
784 
ScheduleMsgAndHandle(Message * inMsg)785 void SingleVerSyncStateMachine::ScheduleMsgAndHandle(Message *inMsg)
786 {
787     dataSync_->PutDataMsg(inMsg);
788     while (true) {
789         bool isNeedHandle = true;
790         bool isNeedContinue = true;
791         Message *msg = dataSync_->MoveNextDataMsg(context_, isNeedHandle, isNeedContinue);
792         if (!isNeedContinue) {
793             break;
794         }
795         if (msg == nullptr) {
796             if (dataSync_->IsNeedReloadQueue()) {
797                 continue;
798             }
799             dataSync_->ScheduleInfoHandle(false, false, nullptr);
800             break;
801         }
802         bool isNeedClearMap = false;
803         if (isNeedHandle) {
804             int errCode = HandleDataRequestRecv(msg);
805             if (context_->IsReceiveWaterMarkErr() || errCode == -E_NEED_ABILITY_SYNC || errCode == -E_NEED_TIME_SYNC) {
806                 isNeedClearMap = true;
807             }
808             if (errCode == -E_TIMEOUT) {
809                 isNeedHandle = false;
810             }
811         } else {
812             dataSync_->SendFinishedDataAck(context_, msg);
813         }
814         if (context_->GetRemoteSoftwareVersion() < SOFTWARE_VERSION_RELEASE_3_0) {
815             // for lower version, no need to handle map schedule info, just reset schedule working status
816             isNeedHandle = false;
817         }
818         dataSync_->ScheduleInfoHandle(isNeedHandle, isNeedClearMap, msg);
819         delete msg;
820     }
821 }
822 
ControlPktRecv(Message * inMsg)823 int SingleVerSyncStateMachine::ControlPktRecv(Message *inMsg)
824 {
825     int errCode = E_OK;
826     switch (inMsg->GetMessageType()) {
827         case TYPE_REQUEST:
828             errCode = dataSync_->ControlCmdRequestRecv(context_, inMsg);
829             break;
830         case TYPE_RESPONSE:
831             errCode = HandleControlAckRecv(inMsg);
832             break;
833         default:
834             errCode = -E_INVALID_ARGS;
835             break;
836     }
837     return errCode;
838 }
839 
StepToTimeout(TimerId timerId)840 void SingleVerSyncStateMachine::StepToTimeout(TimerId timerId)
841 {
842     std::lock_guard<std::mutex> lock(stateMachineLock_);
843     TimerId timer = syncContext_->GetTimerId();
844     if (timer != timerId) {
845         return;
846     }
847     SwitchStateAndStep(Event::TIME_OUT_EVENT);
848 }
849 
850 namespace {
851 struct StateNode {
852     int errCode = 0;
853     SyncOperation::Status status = SyncOperation::OP_WAITING;
854 };
855 }
GetSyncOperationStatus(int errCode) const856 int SingleVerSyncStateMachine::GetSyncOperationStatus(int errCode) const
857 {
858     static const StateNode stateNodes[] = {
859         { -E_SCHEMA_MISMATCH,                 SyncOperation::OP_SCHEMA_INCOMPATIBLE },
860         { -E_EKEYREVOKED,                     SyncOperation::OP_EKEYREVOKED_FAILURE },
861         { -E_SECURITY_OPTION_CHECK_ERROR,     SyncOperation::OP_SECURITY_OPTION_CHECK_FAILURE },
862         { -E_BUSY,                            SyncOperation::OP_BUSY_FAILURE },
863         { -E_NOT_PERMIT,                      SyncOperation::OP_PERMISSION_CHECK_FAILED },
864         { -E_TIMEOUT,                         SyncOperation::OP_TIMEOUT },
865         { -E_INVALID_QUERY_FORMAT,            SyncOperation::OP_QUERY_FORMAT_FAILURE },
866         { -E_INVALID_QUERY_FIELD,             SyncOperation::OP_QUERY_FIELD_FAILURE },
867         { -E_FEEDBACK_UNKNOWN_MESSAGE,        SyncOperation::OP_NOT_SUPPORT },
868         { -E_FEEDBACK_COMMUNICATOR_NOT_FOUND, SyncOperation::OP_COMM_ABNORMAL },
869         { -E_NOT_SUPPORT,                     SyncOperation::OP_NOT_SUPPORT },
870         { -E_INTERCEPT_DATA_FAIL,             SyncOperation::OP_INTERCEPT_DATA_FAIL },
871         { -E_MAX_LIMITS,                      SyncOperation::OP_MAX_LIMITS },
872         { -E_DISTRIBUTED_SCHEMA_CHANGED,      SyncOperation::OP_SCHEMA_CHANGED },
873         { -E_NOT_REGISTER,                    SyncOperation::OP_NOT_SUPPORT },
874         { -E_DENIED_SQL,                      SyncOperation::OP_DENIED_SQL },
875         { -E_REMOTE_OVER_SIZE,                SyncOperation::OP_MAX_LIMITS },
876         { -E_INVALID_PASSWD_OR_CORRUPTED_DB,  SyncOperation::OP_NOTADB_OR_CORRUPTED }
877     };
878     const auto &result = std::find_if(std::begin(stateNodes), std::end(stateNodes), [errCode](const auto &node) {
879         return node.errCode == errCode;
880     });
881     return result == std::end(stateNodes) ? SyncOperation::OP_FAILED : static_cast<int>((*result).status);
882 }
883 
TimeMarkSyncRecv(const Message * inMsg)884 int SingleVerSyncStateMachine::TimeMarkSyncRecv(const Message *inMsg)
885 {
886     LOGD("[StateMachine][TimeMarkSyncRecv] type=%d,label=%s,dev=%s", inMsg->GetMessageType(),
887         dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
888     {
889         std::lock_guard<std::mutex> lock(stateMachineLock_);
890         (void)ResetWatchDog();
891     }
892     if (inMsg->GetMessageType() == TYPE_REQUEST) {
893         return timeSync_->RequestRecv(inMsg);
894     } else if (inMsg->GetMessageType() == TYPE_RESPONSE) {
895         int errCode = timeSync_->AckRecv(inMsg, context_->GetRequestSessionId());
896         if (errCode != E_OK) {
897             LOGE("[StateMachine][TimeMarkSyncRecv] AckRecv failed errCode=%d", errCode);
898             if (inMsg->GetSessionId() != 0 && inMsg->GetSessionId() == context_->GetRequestSessionId()) {
899                 context_->SetTaskErrCode(errCode);
900                 InnerErrorAbort(inMsg->GetSessionId());
901             }
902             return errCode;
903         }
904         std::lock_guard<std::mutex> lock(stateMachineLock_);
905         SwitchStateAndStep(TIME_SYNC_FINISHED_EVENT);
906         return E_OK;
907     } else {
908         return -E_INVALID_ARGS;
909     }
910 }
911 
Clear()912 void SingleVerSyncStateMachine::Clear()
913 {
914     dataSync_ = nullptr;
915     timeSync_ = nullptr;
916     abilitySync_ = nullptr;
917     context_ = nullptr;
918     syncInterface_ = nullptr;
919 }
920 
IsPacketValid(const Message * inMsg) const921 bool SingleVerSyncStateMachine::IsPacketValid(const Message *inMsg) const
922 {
923     if (inMsg == nullptr) {
924         return false;
925     }
926 
927     if ((inMsg->GetMessageId() < TIME_SYNC_MESSAGE) || (inMsg->GetMessageId() >= UNKNOW_MESSAGE)) {
928         LOGE("[StateMachine][IsPacketValid] Message is invalid, id=%d", inMsg->GetMessageId());
929         return false;
930     }
931     // filter invalid ack at first
932     bool isResponseType = (inMsg->GetMessageType() == TYPE_RESPONSE);
933     if (isResponseType && (inMsg->GetMessageId() == CONTROL_SYNC_MESSAGE) &&
934         (inMsg->GetSessionId() != context_->GetRequestSessionId())) {
935         LOGE("[StateMachine][IsPacketValid] Control Message is invalid, label=%s, dev=%s",
936             dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
937         return false;
938     }
939     if (isResponseType && (inMsg->GetMessageId() != TIME_SYNC_MESSAGE) &&
940         (inMsg->GetSessionId() != context_->GetRequestSessionId()) &&
941         (inMsg->GetSessionId() != context_->GetResponseSessionId())) {
942         LOGE("[StateMachine][IsPacketValid] Data Message is invalid, label=%s, dev=%s",
943             dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
944         return false;
945     }
946 
947     // timeSync and abilitySync don't need to check sequenceId
948     if (inMsg->GetMessageId() == TIME_SYNC_MESSAGE || inMsg->GetMessageId() == ABILITY_SYNC_MESSAGE ||
949         inMsg->GetMessageId() == CONTROL_SYNC_MESSAGE) {
950         return true;
951     }
952     // sequenceId will be checked in dataSync
953     return true;
954 }
955 
PreStartPullResponse()956 void SingleVerSyncStateMachine::PreStartPullResponse()
957 {
958     SingleVerSyncTarget target;
959     context_->PopResponseTarget(target);
960     context_->SetEndMark(target.GetEndWaterMark());
961     context_->SetResponseSessionId(target.GetResponseSessionId());
962     context_->SetMode(SyncModeType::RESPONSE_PULL);
963     context_->ReSetSequenceId();
964     context_->SetQuerySync(target.IsQuerySync());
965     context_->SetQuery(target.GetQuery());
966 }
967 
CheckIsStartPullResponse() const968 bool SingleVerSyncStateMachine::CheckIsStartPullResponse() const
969 {
970     // Other state will step to do pull response, only this statem we need to step the statemachine
971     if (currentState_ == WAIT_FOR_RECEIVE_DATA_FINISH) {
972         return true;
973     }
974     return false;
975 }
976 
MessageCallbackPre(const Message * inMsg)977 int SingleVerSyncStateMachine::MessageCallbackPre(const Message *inMsg)
978 {
979     RefObject::AutoLock lock(context_);
980     if (context_->IsKilled()) {
981         return -E_OBJ_IS_KILLED;
982     }
983 
984     if (!IsPacketValid(inMsg)) {
985         return -E_INVALID_ARGS;
986     }
987     return E_OK;
988 }
989 
AddPullResponseTarget(const Message * inMsg,WaterMark pullEndWatermark)990 void SingleVerSyncStateMachine::AddPullResponseTarget(const Message *inMsg, WaterMark pullEndWatermark)
991 {
992     int messageType = static_cast<int>(inMsg->GetMessageId());
993     uint32_t sessionId = inMsg->GetSessionId();
994     if (pullEndWatermark == 0) {
995         LOGE("[StateMachine][AddPullResponseTarget] pullEndWatermark is 0!");
996         return;
997     }
998     if (context_->GetResponseSessionId() == sessionId || context_->FindResponseSyncTarget(sessionId)) {
999         LOGI("[StateMachine][AddPullResponseTarget] task is already running");
1000         return;
1001     }
1002     const DataRequestPacket *packet = inMsg->GetObject<DataRequestPacket>();
1003     if (packet == nullptr) {
1004         LOGE("[AddPullResponseTarget] get packet object failed");
1005         return;
1006     }
1007     SingleVerSyncTarget *targetTmp = new (std::nothrow) SingleVerSyncTarget;
1008     if (targetTmp == nullptr) {
1009         LOGE("[StateMachine][AddPullResponseTarget] add failed, may oom");
1010         return;
1011     }
1012     targetTmp->SetTaskType(ISyncTarget::RESPONSE);
1013     if (messageType == QUERY_SYNC_MESSAGE) {
1014         targetTmp->SetQuery(packet->GetQuery());
1015         targetTmp->SetQuerySync(true);
1016     }
1017     targetTmp->SetMode(SyncModeType::RESPONSE_PULL);
1018     targetTmp->SetEndWaterMark(pullEndWatermark);
1019     targetTmp->SetResponseSessionId(sessionId);
1020     if (context_->AddSyncTarget(targetTmp) != E_OK) {
1021         delete targetTmp;
1022         return;
1023     }
1024     if (CheckIsStartPullResponse()) {
1025         SwitchStateAndStep(TransformErrCodeToEvent(-E_NEED_PULL_REPONSE));
1026     }
1027 }
1028 
TransformErrCodeToEvent(int errCode) const1029 Event SingleVerSyncStateMachine::TransformErrCodeToEvent(int errCode) const
1030 {
1031     switch (errCode) {
1032         case -E_TIMEOUT:
1033             return TransforTimeOutErrCodeToEvent();
1034         case -static_cast<int>(VERSION_NOT_SUPPOR_EVENT):
1035             return Event::VERSION_NOT_SUPPOR_EVENT;
1036         case -E_SEND_DATA:
1037             return Event::SEND_DATA_EVENT;
1038         case -E_NO_DATA_SEND:
1039             return Event::SEND_FINISHED_EVENT;
1040         case -E_RECV_FINISHED:
1041             return Event::RECV_FINISHED_EVENT;
1042         case -E_NEED_ABILITY_SYNC:
1043             return Event::NEED_ABILITY_SYNC_EVENT;
1044         case -E_NO_SYNC_TASK:
1045             return Event::ALL_TASK_FINISHED_EVENT;
1046         case -E_NEED_PULL_REPONSE:
1047             return Event::START_PULL_RESPONSE_EVENT;
1048         case -E_RE_SEND_DATA:
1049             return Event::RE_SEND_DATA_EVENT;
1050         case -E_NEED_TIME_SYNC:
1051             return Event::NEED_TIME_SYNC_EVENT;
1052         default:
1053             return Event::INNER_ERR_EVENT;
1054     }
1055 }
1056 
IsNeedResetWatchdog(const Message * inMsg) const1057 bool SingleVerSyncStateMachine::IsNeedResetWatchdog(const Message *inMsg) const
1058 {
1059     if (inMsg == nullptr) {
1060         return false;
1061     }
1062 
1063     if (IsNeedErrCodeHandle(inMsg->GetSessionId())) {
1064         return true;
1065     }
1066 
1067     int msgType = inMsg->GetMessageType();
1068     if (msgType == TYPE_RESPONSE || msgType == TYPE_NOTIFY) {
1069         if (inMsg->GetSessionId() == context_->GetResponseSessionId()) {
1070             // Pull response ack also should reset watchdog
1071             return true;
1072         }
1073     }
1074 
1075     return false;
1076 }
1077 
TransforTimeOutErrCodeToEvent() const1078 Event SingleVerSyncStateMachine::TransforTimeOutErrCodeToEvent() const
1079 {
1080     if (syncContext_->IsSyncTaskNeedRetry() && (syncContext_->GetRetryTime() < syncContext_->GetSyncRetryTimes())) {
1081         return Event::WAIT_TIME_OUT_EVENT;
1082     } else {
1083         return Event::TIME_OUT_EVENT;
1084     }
1085 }
1086 
IsNeedErrCodeHandle(uint32_t sessionId) const1087 bool SingleVerSyncStateMachine::IsNeedErrCodeHandle(uint32_t sessionId) const
1088 {
1089     // omit to set sessionId so version_3 should skip to compare sessionid.
1090     if (sessionId == context_->GetRequestSessionId() ||
1091         context_->GetRemoteSoftwareVersion() == SOFTWARE_VERSION_RELEASE_2_0) {
1092         return true;
1093     }
1094     return false;
1095 }
1096 
PushPullDataRequestEvokeErrHandle()1097 void SingleVerSyncStateMachine::PushPullDataRequestEvokeErrHandle()
1098 {
1099     // the pushpull sync task should wait for send finished after remote dev get data occur E_EKEYREVOKED error.
1100     if (context_->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_2_0 &&
1101         SyncOperation::TransferSyncMode(context_->GetMode()) == SyncModeType::PUSH_AND_PULL) { // LCOV_EXCL_BR_LINE
1102         LOGI("data request errCode = %d, wait for send finished", -E_EKEYREVOKED);
1103         context_->SetTaskErrCode(-E_EKEYREVOKED);
1104         context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
1105         SwitchStateAndStep(Event::RECV_FINISHED_EVENT);
1106     } else {
1107         context_->SetTaskErrCode(-E_EKEYREVOKED);
1108         SwitchStateAndStep(Event::INNER_ERR_EVENT);
1109     }
1110 }
1111 
DataRecvErrCodeHandle(uint32_t sessionId,int errCode)1112 void SingleVerSyncStateMachine::DataRecvErrCodeHandle(uint32_t sessionId, int errCode)
1113 {
1114     if (IsNeedErrCodeHandle(sessionId)) {
1115         switch (errCode) {
1116             case E_OK:
1117                 break;
1118             case -E_NOT_PERMIT:
1119                 context_->SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
1120                 break;
1121             case -E_RECV_FINISHED:
1122                 context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
1123                 SwitchStateAndStep(Event::RECV_FINISHED_EVENT);
1124                 break;
1125             case -E_EKEYREVOKED:
1126                 PushPullDataRequestEvokeErrHandle();
1127                 break;
1128             case -E_BUSY:
1129             case -E_DISTRIBUTED_SCHEMA_CHANGED:
1130             case -E_DISTRIBUTED_SCHEMA_NOT_FOUND:
1131             case -E_FEEDBACK_COMMUNICATOR_NOT_FOUND:
1132             case -E_FEEDBACK_UNKNOWN_MESSAGE:
1133             case -E_INTERCEPT_DATA_FAIL:
1134             case -E_INVALID_PASSWD_OR_CORRUPTED_DB:
1135             case -E_INVALID_QUERY_FIELD:
1136             case -E_INVALID_QUERY_FORMAT:
1137             case -E_MAX_LIMITS:
1138             case -E_NOT_REGISTER:
1139             case -E_NOT_SUPPORT:
1140             case -E_SECURITY_OPTION_CHECK_ERROR:
1141                 context_->SetTaskErrCode(errCode);
1142                 SwitchStateAndStep(Event::INNER_ERR_EVENT);
1143                 break;
1144             default:
1145                 SwitchStateAndStep(Event::INNER_ERR_EVENT);
1146                 break;
1147         }
1148     }
1149 }
1150 
AbilityMsgSessionIdCheck(const Message * inMsg)1151 bool SingleVerSyncStateMachine::AbilityMsgSessionIdCheck(const Message *inMsg)
1152 {
1153     if (inMsg != nullptr && inMsg->GetSessionId() == context_->GetRequestSessionId()) { // LCOV_EXCL_BR_LINE
1154         return true;
1155     }
1156     LOGE("[AbilitySync] session check failed,dev=%s", STR_MASK(context_->GetDeviceId()));
1157     return false;
1158 }
1159 
GetSyncType(uint32_t messageId) const1160 SyncType SingleVerSyncStateMachine::GetSyncType(uint32_t messageId) const
1161 {
1162     if (messageId == QUERY_SYNC_MESSAGE) { // LCOV_EXCL_BR_LINE
1163         return SyncType::QUERY_SYNC_TYPE;
1164     }
1165     return SyncType::MANUAL_FULL_SYNC_TYPE;
1166 }
1167 
DataAckRecvErrCodeHandle(int errCode,bool handleError)1168 void SingleVerSyncStateMachine::DataAckRecvErrCodeHandle(int errCode, bool handleError)
1169 {
1170     switch (errCode) {
1171         case -E_NEED_ABILITY_SYNC:
1172             NeedAbilitySyncHandle();
1173             break;
1174         case -E_NEED_TIME_SYNC:
1175             timeSync_->ClearTimeSyncFinish();
1176             break;
1177         case -E_NOT_PERMIT:
1178             if (handleError) {
1179                 context_->SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
1180             }
1181             break;
1182         case -E_BUSY:
1183         case -E_DISTRIBUTED_SCHEMA_CHANGED:
1184         case -E_DISTRIBUTED_SCHEMA_NOT_FOUND:
1185         case -E_EKEYREVOKED:
1186         case -E_FEEDBACK_COMMUNICATOR_NOT_FOUND:
1187         case -E_FEEDBACK_UNKNOWN_MESSAGE:
1188         case -E_INTERCEPT_DATA_FAIL:
1189         case -E_INVALID_PASSWD_OR_CORRUPTED_DB:
1190         case -E_INVALID_QUERY_FIELD:
1191         case -E_INVALID_QUERY_FORMAT:
1192         case -E_MAX_LIMITS:
1193         case -E_NOT_REGISTER:
1194         case -E_NOT_SUPPORT:
1195         case -E_SECURITY_OPTION_CHECK_ERROR:
1196             if (handleError) {
1197                 context_->SetTaskErrCode(errCode);
1198             }
1199             break;
1200         default:
1201             break;
1202     }
1203 }
1204 
IsNeedTriggerQueryAutoSync(Message * inMsg,QuerySyncObject & query)1205 bool SingleVerSyncStateMachine::IsNeedTriggerQueryAutoSync(Message *inMsg, QuerySyncObject &query)
1206 {
1207     return SingleVerDataSyncUtils::IsNeedTriggerQueryAutoSync(inMsg, query);
1208 }
1209 
JumpStatusAfterAbilitySync(int mode)1210 void SingleVerSyncStateMachine::JumpStatusAfterAbilitySync(int mode)
1211 {
1212     if ((mode == SyncModeType::SUBSCRIBE_QUERY) || (mode == SyncModeType::UNSUBSCRIBE_QUERY)) {
1213         SwitchStateAndStep(CONTROL_CMD_EVENT);
1214     } else {
1215         SwitchStateAndStep(ABILITY_SYNC_FINISHED_EVENT);
1216     }
1217 }
1218 
ControlAckRecvErrCodeHandle(int errCode)1219 void SingleVerSyncStateMachine::ControlAckRecvErrCodeHandle(int errCode)
1220 {
1221     switch (errCode) {
1222         case -E_NEED_ABILITY_SYNC:
1223             NeedAbilitySyncHandle();
1224             break;
1225         case -E_NO_DATA_SEND:
1226             context_->SetOperationStatus(SyncOperation::OP_SEND_FINISHED);
1227             break;
1228         case -E_NOT_PERMIT:
1229             context_->SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
1230             break;
1231         // other errCode use default
1232         default:
1233             context_->SetTaskErrCode(errCode);
1234             break;
1235     }
1236 }
1237 
GetLocalWaterMark(const DeviceID & deviceId,uint64_t & outValue)1238 void SingleVerSyncStateMachine::GetLocalWaterMark(const DeviceID &deviceId, uint64_t &outValue)
1239 {
1240     metadata_->GetLocalWaterMark(deviceId, outValue);
1241 }
1242 
GetSendQueryWaterMark(const std::string & queryId,const DeviceID & deviceId,bool isAutoLift,uint64_t & outValue)1243 int SingleVerSyncStateMachine::GetSendQueryWaterMark(const std::string &queryId,  const DeviceID &deviceId,
1244     bool isAutoLift, uint64_t &outValue)
1245 {
1246     return metadata_->GetSendQueryWaterMark(queryId, deviceId, outValue, isAutoLift);
1247 }
1248 
ResponsePullError(int errCode,bool ignoreInnerErr)1249 void SingleVerSyncStateMachine::ResponsePullError(int errCode, bool ignoreInnerErr)
1250 {
1251     Event event = TransformErrCodeToEvent(errCode);
1252     if (event == INNER_ERR_EVENT) {
1253         if (ignoreInnerErr) {
1254             event = RESPONSE_TASK_FINISHED_EVENT;
1255         } else if (context_ != nullptr) {
1256             context_->SetTaskErrCode(errCode);
1257         }
1258     }
1259     SwitchStateAndStep(event);
1260 }
1261 
InnerErrorAbort(uint32_t sessionId)1262 void SingleVerSyncStateMachine::InnerErrorAbort(uint32_t sessionId)
1263 {
1264     std::lock_guard<std::mutex> lock(stateMachineLock_);
1265     uint32_t requestSessionId = context_->GetRequestSessionId();
1266     if (sessionId != requestSessionId) {
1267         LOGD("[SingleVerSyncStateMachine][InnerErrorAbort] Ignore abort by different sessionId");
1268         return;
1269     }
1270     if (SwitchMachineState(Event::INNER_ERR_EVENT) == E_OK) {
1271         SyncStep();
1272     }
1273 }
1274 
NotifyClosing()1275 void SingleVerSyncStateMachine::NotifyClosing()
1276 {
1277     if (timeSync_ != nullptr) {
1278         timeSync_->Close();
1279     }
1280 }
1281 
AbilitySyncNotifyRecv(const Message * inMsg)1282 int SingleVerSyncStateMachine::AbilitySyncNotifyRecv(const Message *inMsg)
1283 {
1284     const AbilitySyncAckPacket *packet = inMsg->GetObject<AbilitySyncAckPacket>();
1285     if (packet == nullptr) {
1286         return -E_INVALID_ARGS;
1287     }
1288     int ackCode = packet->GetAckCode();
1289     if (ackCode != AbilitySync::CHECK_SUCCESS && ackCode != AbilitySync::LAST_NOTIFY) {
1290         LOGE("[StateMachine][AbilitySyncRecv] ackCode check failed,ackCode=%d", ackCode);
1291         context_->SetTaskErrCode(ackCode);
1292         std::lock_guard<std::mutex> lock(stateMachineLock_);
1293         SwitchStateAndStep(Event::INNER_ERR_EVENT);
1294         return E_OK;
1295     }
1296     if (ackCode == AbilitySync::LAST_NOTIFY && AbilityMsgSessionIdCheck(inMsg)) {
1297         abilitySync_->SetAbilitySyncFinishedStatus(true, *context_);
1298         // while recv last notify means ability sync finished,it is better to reset watchDog to avoid timeout.
1299         LOGI("[StateMachine][AbilitySyncRecv] ability sync finished,label=%s,dev=%s",
1300             dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
1301         context_->SetRemoteSoftwareVersion(packet->GetSoftwareVersion());
1302         currentRemoteVersionId_ = context_->GetRemoteSoftwareVersionId();
1303         std::lock_guard<std::mutex> lock(stateMachineLock_);
1304         (void)ResetWatchDog();
1305         JumpStatusAfterAbilitySync(context_->GetMode());
1306     } else if (ackCode != AbilitySync::LAST_NOTIFY) {
1307         abilitySync_->AckNotifyRecv(inMsg, context_);
1308     }
1309     return E_OK;
1310 }
1311 
SchemaChange()1312 void SingleVerSyncStateMachine::SchemaChange()
1313 {
1314     abilitySync_->SetAbilitySyncFinishedStatus(false, *context_);
1315 }
1316 
TimeChange()1317 void SingleVerSyncStateMachine::TimeChange()
1318 {
1319     if (timeSync_ == nullptr) {
1320         LOGW("[SingleVerSyncStateMachine] time sync is null when time change");
1321         return;
1322     }
1323     timeSync_->ClearTimeSyncFinish();
1324 }
1325 } // namespace DistributedDB
1326