From 37078e197740817fa4702269a773b495b3e9964a Mon Sep 17 00:00:00 2001 From: zhoushengchen <3057083327@qq.com> Date: Sat, 29 Aug 2026 11:10:38 +0800 Subject: [PATCH] fix: use stable cursor for ding payable workflow sync --- .../DingPayableWorkflowMirrorSynchronizer.php | 139 ++++++++++-- ...gPayableWorkflowMirrorSynchronizerTest.php | 199 ++++++++++++++++++ 2 files changed, 318 insertions(+), 20 deletions(-) create mode 100644 tests/Unit/EcommerceFinance/DingPayableWorkflowMirrorSynchronizerTest.php diff --git a/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php b/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php index dcef66f7..167b2894 100644 --- a/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php @@ -5,6 +5,7 @@ namespace Modules\EcommerceFinance\Services; use Illuminate\Database\ConnectionInterface; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\Schema; +use Illuminate\Support\Carbon; use Illuminate\Support\Str; use RuntimeException; use Throwable; @@ -49,29 +50,28 @@ class DingPayableWorkflowMirrorSynchronizer } $mode = $full ? 'full' : 'incremental'; + $stateCursor = $this->decodeJson($state['cursor'] ?? null); + $highWatermark = $this->cursorFromState($this->decodeJson($state['high_watermark'] ?? null)); $resume = is_array($state) && ($state['run_mode'] ?? null) === $mode && in_array($state['status'] ?? null, ['running', 'failed'], true) && is_string($state['generation'] ?? null) - && is_array($state['cursor'] ?? null); + && $this->isCompatibleCursor($stateCursor); $generation = $resume ? (string) $state['generation'] : (string) Str::uuid(); - $cursor = $resume - ? (string) ($state['cursor']['processInstanceId'] ?? '') - : ($full - ? '' - : (string) ($this->decodeJson($state['high_watermark'] ?? null)['processInstanceId'] ?? '')); + [$windowStart, $windowEnd] = $this->window($state, $mode, $resume); + $cursor = $resume ? $this->cursorFromState($stateCursor) : null; $rows = $resume ? (int) ($state['last_upserted_rows'] ?? 0) : 0; $batchSize = min(max($batchSize ?? (int) ($config['batch_size'] ?? 100), 1), 500); $maxBatches = max($maxBatches ?? (int) ($config['max_batches_per_run'] ?? 2), 0); - $this->start($dataset, $target, $source, $mode, $generation, $cursor, $resume); + $this->start($dataset, $target, $source, $mode, $generation, $resume, $windowStart, $windowEnd); $batches = 0; try { while (true) { - $workflows = $this->readWorkflows($source, $processCode, $cursor, $batchSize); + $workflows = $this->readWorkflows($source, $processCode, $cursor, $windowStart, $windowEnd, $batchSize); if ($workflows === []) { - $this->complete($dataset, $target, $mode, $generation, $rows, $cursor); + $this->complete($dataset, $target, $mode, $generation, $rows, $cursor ?? $highWatermark); return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset]; } @@ -116,7 +116,11 @@ class DingPayableWorkflowMirrorSynchronizer ]; } - $cursor = (string) ($workflows[array_key_last($workflows)]['processInstanceId'] ?? $cursor); + $nextCursor = $this->cursorFromWorkflow($workflows[array_key_last($workflows)]); + if ($nextCursor === null) { + throw new RuntimeException('ding_payable_workflow_missing_watermark'); + } + $cursor = $nextCursor; $rows += count($payloads); $batches++; $this->connection->transaction(function () use ($dataset, $target, $payloads, $cursor, $generation, $rows, $now): void { @@ -130,7 +134,9 @@ class DingPayableWorkflowMirrorSynchronizer $this->connection->table('business_mirror_table_states') ->where('dataset_key', $dataset) ->update([ - 'cursor' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE), + 'cursor' => $cursor === null + ? null + : json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES), 'last_upserted_rows' => $rows, 'updated_at' => $now, ]); @@ -184,7 +190,16 @@ class DingPayableWorkflowMirrorSynchronizer return $state; } - private function start(string $dataset, string $target, string $source, string $mode, string $generation, string $cursor, bool $resume): void + private function start( + string $dataset, + string $target, + string $source, + string $mode, + string $generation, + bool $resume, + ?Carbon $windowStart, + ?Carbon $windowEnd, + ): void { $now = now(); $values = [ @@ -199,7 +214,12 @@ class DingPayableWorkflowMirrorSynchronizer 'updated_at' => $now, ]; if (! $resume) { - $values += ['cursor' => null, 'window_started_at' => null, 'window_ends_at' => null, 'last_upserted_rows' => 0]; + $values += [ + 'cursor' => null, + 'window_started_at' => $windowStart, + 'window_ends_at' => $windowEnd, + 'last_upserted_rows' => 0, + ]; } $query = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset); $query->exists() @@ -207,7 +227,7 @@ class DingPayableWorkflowMirrorSynchronizer : $this->connection->table('business_mirror_table_states')->insert(['dataset_key' => $dataset, 'created_at' => $now] + $values); } - private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, string $cursor): void + private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, ?array $cursor): void { $now = now(); $this->connection->transaction(function () use ($dataset, $target, $mode, $generation, $rows, $cursor, $now): void { @@ -219,7 +239,9 @@ class DingPayableWorkflowMirrorSynchronizer $values = [ 'status' => 'succeeded', 'cursor' => null, - 'high_watermark' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE), + 'high_watermark' => $cursor === null + ? null + : json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES), 'last_succeeded_at' => $now, 'last_upserted_rows' => $rows, 'mirror_row_count' => $this->connection->table($target)->count(), @@ -248,20 +270,97 @@ class DingPayableWorkflowMirrorSynchronizer } /** @return list> */ - private function readWorkflows(string $source, string $processCode, string $cursor, int $limit): array + private function readWorkflows( + string $source, + string $processCode, + ?array $cursor, + ?Carbon $windowStart, + ?Carbon $windowEnd, + int $limit, + ): array { + $watermark = 'COALESCE([updateAt], [createAt])'; $where = '[processCode] = ?'; $bindings = [$processCode]; - if ($cursor !== '') { - $where .= ' AND [processInstanceId] > ?'; - $bindings[] = $cursor; + if ($windowStart !== null && $windowEnd !== null) { + $where .= ' AND ' . $watermark . ' >= ? AND ' . $watermark . ' < ?'; + $bindings[] = $this->sqlServerDateTime($windowStart); + $bindings[] = $this->sqlServerDateTime($windowEnd); + } + if ($cursor !== null) { + $where .= ' AND (' . $watermark . ' > ? OR (' . $watermark . ' = ? AND [processInstanceId] > ?))'; + $bindings[] = (string) ($cursor['updateAt'] ?? ''); + $bindings[] = (string) ($cursor['updateAt'] ?? ''); + $bindings[] = (string) ($cursor['processInstanceId'] ?? ''); } return $this->queries->forSource($source)->select( - 'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId],[joinAt],[updateAt],[createAt],[processCode],[flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY [processInstanceId]', + 'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId], [joinAt], COALESCE([updateAt], [createAt]) AS [updateAt], [createAt], [processCode], [flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY ' . $watermark . ', [processInstanceId]', $bindings, ); } + /** @return array{0:?Carbon,1:?Carbon} */ + private function window(?array $state, string $mode, bool $resume): array + { + if ($mode === 'full') { + return [null, null]; + } + + if ($resume && ($state['window_started_at'] ?? null) !== null && ($state['window_ends_at'] ?? null) !== null) { + return [ + Carbon::parse((string) $state['window_started_at']), + Carbon::parse((string) $state['window_ends_at']), + ]; + } + + $overlap = max((int) config('upstream_mirrors.table_sync.overlap_hours', 48), 1); + $lastSucceeded = ($state['last_succeeded_at'] ?? null) !== null + ? Carbon::parse((string) $state['last_succeeded_at']) + : now(); + + return [$lastSucceeded->copy()->subHours($overlap), now()->addSecond()]; + } + + /** @return array{updateAt:string,processInstanceId:string}|null */ + private function cursorFromState(array $cursor): ?array + { + if (($cursor['updateAt'] ?? '') === '' && ($cursor['processInstanceId'] ?? '') === '') { + return null; + } + + if (($cursor['updateAt'] ?? '') === '') { + return null; + } + + return [ + 'updateAt' => (string) $cursor['updateAt'], + 'processInstanceId' => (string) ($cursor['processInstanceId'] ?? ''), + ]; + } + + /** @param array $workflow @return array{updateAt:string,processInstanceId:string}|null */ + private function cursorFromWorkflow(array $workflow): ?array + { + $id = trim((string) ($workflow['processInstanceId'] ?? '')); + $updateAt = trim((string) ($workflow['updateAt'] ?? $workflow['createAt'] ?? '')); + + if ($id === '' || $updateAt === '') { + return null; + } + + return ['updateAt' => $updateAt, 'processInstanceId' => $id]; + } + + private function isCompatibleCursor(array $cursor): bool + { + return $cursor === [] || array_key_exists('updateAt', $cursor); + } + + private function sqlServerDateTime(Carbon $value): string + { + return $value->format('Y-m-d\TH:i:s.v'); + } + /** @param list $ids @param list $columns @return list> */ private function readRelated(string $source, string $table, array $columns, array $ids, string $order): array { diff --git a/tests/Unit/EcommerceFinance/DingPayableWorkflowMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/DingPayableWorkflowMirrorSynchronizerTest.php new file mode 100644 index 00000000..0018d5d6 --- /dev/null +++ b/tests/Unit/EcommerceFinance/DingPayableWorkflowMirrorSynchronizerTest.php @@ -0,0 +1,199 @@ + [ + 'driver' => 'sqlite', + 'database' => ':memory:', + 'prefix' => '', + ], + 'upstream_mirrors.connection' => 'business_mirror', + 'upstream_mirrors.table_sync.overlap_hours' => 48, + 'upstream_mirrors.ding_payable_workflows' => [ + 'dataset_key' => 'operations_replica.rpt.ding_payable_workflows', + 'source_key' => 'operations_replica', + 'process_code' => 'PROC-TEST', + 'target_table' => 'ding_payable_workflow_mirror', + 'batch_size' => 100, + 'max_batches_per_run' => 2, + ], + ]); + DB::purge('business_mirror'); + + Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $table): void { + $table->string('dataset_key')->primary(); + $table->string('source_key'); + $table->string('source_object'); + $table->string('target_table'); + $table->string('status')->default('pending'); + $table->string('run_mode')->nullable(); + $table->string('generation')->nullable(); + $table->json('cursor')->nullable(); + $table->json('high_watermark')->nullable(); + $table->dateTime('window_started_at')->nullable(); + $table->dateTime('window_ends_at')->nullable(); + $table->dateTime('initial_completed_at')->nullable(); + $table->dateTime('last_full_succeeded_at')->nullable(); + $table->dateTime('last_attempted_at')->nullable(); + $table->dateTime('last_succeeded_at')->nullable(); + $table->unsignedBigInteger('last_upserted_rows')->default(0); + $table->unsignedBigInteger('mirror_row_count')->default(0); + $table->string('last_error_code')->nullable(); + $table->timestamps(); + }); + + $this->query = new FakeDingPayableWorkflowQuery(); + $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory { + public function __construct(private readonly UpstreamReadOnlyQuery $query) {} + + public function forSource(string $source): UpstreamReadOnlyQuery + { + return $this->query; + } + }; + $this->synchronizer = new DingPayableWorkflowMirrorSynchronizer($factory); + } + + public function test_full_sync_resumes_with_update_time_and_process_id_tie_breaker(): void + { + $sameUpdateAt = '2026-08-29 10:00:00.1234567'; + $this->query->workflowBatches = [[ + $this->workflow('A-process', $sameUpdateAt), + $this->workflow('B-process', $sameUpdateAt), + ]]; + + $result = $this->synchronizer->sync(true, 2, 1); + + $this->assertSame('running', $result['status']); + $this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $this->query->calls[0]['sql']); + $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first(); + $this->assertSame([ + 'updateAt' => $sameUpdateAt, + 'processInstanceId' => 'B-process', + ], json_decode((string) $state->cursor, true)); + + $this->query->calls = []; + $this->query->workflowBatches = [[ + $this->workflow('C-process', $sameUpdateAt), + ]]; + + $result = $this->synchronizer->sync(true, 2, 0); + + $this->assertSame('succeeded', $result['status']); + $this->assertStringContainsString( + 'COALESCE([updateAt], [createAt]) > ? OR (COALESCE([updateAt], [createAt]) = ? AND [processInstanceId] > ?', + $this->query->calls[0]['sql'], + ); + $this->assertSame([$sameUpdateAt, $sameUpdateAt, 'B-process'], array_slice($this->query->calls[0]['bindings'], -3)); + $this->assertSame(3, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count()); + } + + public function test_incremental_sync_uses_overlap_window_and_keeps_fixed_process_code(): void + { + $this->query->workflowBatches = [[ + $this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')), + ]]; + $this->synchronizer->sync(true, 10, 0); + + $this->query->calls = []; + $this->query->workflowBatches = [[ + $this->workflow('new-process', now()->format('Y-m-d H:i:s.u'), 'RUNNING'), + ]]; + + $result = $this->synchronizer->sync(false, 10, 0); + + $this->assertSame('succeeded', $result['status']); + $sql = $this->query->calls[0]['sql']; + $this->assertStringContainsString('[processCode] = ?', $sql); + $this->assertStringContainsString('COALESCE([updateAt], [createAt]) >= ? AND COALESCE([updateAt], [createAt]) < ?', $sql); + $this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $sql); + $this->assertCount(3, $this->query->calls[0]['bindings']); + $this->assertSame('PROC-TEST', $this->query->calls[0]['bindings'][0]); + $this->assertSame('RUNNING', json_decode( + (string) DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->where('process_instance_id', 'new-process')->value('workflow_payload'), + true, + )['status']); + + $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first(); + $this->assertArrayHasKey('updateAt', json_decode((string) $state->high_watermark, true)); + } + + public function test_legacy_process_id_only_watermark_does_not_skip_new_processes(): void + { + $this->query->workflowBatches = [[ + $this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')), + ]]; + $this->synchronizer->sync(true, 10, 0); + + DB::connection('business_mirror')->table('business_mirror_table_states')->update([ + 'high_watermark' => json_encode(['processInstanceId' => 'z-last-process']), + ]); + $this->query->calls = []; + $this->query->workflowBatches = [[ + $this->workflow('a-new-process', now()->format('Y-m-d H:i:s.u')), + ]]; + + $this->synchronizer->sync(false, 10, 0); + + $this->assertStringNotContainsString('AND [processInstanceId] > ?', $this->query->calls[0]['sql']); + $this->assertSame(2, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count()); + } + + /** @return array */ + private function workflow(string $id, string $updateAt, string $status = 'COMPLETED'): array + { + return [ + 'processInstanceId' => $id, + 'processCode' => 'PROC-TEST', + 'updateAt' => $updateAt, + 'createAt' => $updateAt, + 'status' => $status, + 'originatorUserId' => '', + ]; + } +} + +class FakeDingPayableWorkflowQuery extends UpstreamReadOnlyQuery +{ + /** @var list>> */ + public array $workflowBatches = []; + + /** @var list */ + public array $calls = []; + + public function __construct() + { + parent::__construct('operations_replica'); + } + + public function select(string $sql, array $bindings = []): array + { + $this->calls[] = compact('sql', 'bindings'); + + if (str_contains($sql, 'FROM [rpt].[ding_Workflow]')) { + return array_shift($this->workflowBatches) ?? []; + } + + return []; + } +} -- 2.47.2