From 809bca2e5c790444660df243e38fca4ffbe10f43 Mon Sep 17 00:00:00 2001 From: WuYongJie <549727955@qq.com> Date: Sat, 15 Aug 2026 05:03:19 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=B8=9A=E5=8A=A1=E9=95=9C?= =?UTF-8?q?=E5=83=8F=E5=A2=9E=E9=87=8F=E5=88=86=E9=A1=B5=E8=B6=85=E6=97=B6?= =?UTF-8?q?=20(#61)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Services/BusinessMirrorSynchronizer.php | 26 ++++++++------- .../BusinessMirrorSynchronizerTest.php | 32 ++++++++++++++++++- 2 files changed, 45 insertions(+), 13 deletions(-) diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php index 258bb831..7577037c 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php @@ -23,6 +23,10 @@ class BusinessMirrorSynchronizer public function sync(string $datasetKey, bool $full = false, ?int $batchSize = null, ?int $maxBatches = null): array { $dataset = $this->datasets->get($datasetKey); + // A long-running scheduler may have retained a timed-out PDO instance + // from the previous dataset. Each dataset starts with a fresh upstream + // connection so one source failure cannot poison subsequent mirrors. + UpstreamOdbcConnection::reset($dataset->sourceKey()); $state = $this->mirror->state($dataset); if (! $full && ($state['initial_completed_at'] ?? null) === null) { @@ -77,7 +81,7 @@ class BusinessMirrorSynchronizer return $this->result($dataset, 'succeeded', $mode, $batches, $rows); } - $cursor = $this->cursor($dataset, $batch[array_key_last($batch)], $mode); + $cursor = $this->cursor($dataset, $batch[array_key_last($batch)]); $rows += count($batch); $batches++; $this->mirror->storeBatch($dataset, $columns, $this->normalize($columns, $batch), $cursor, $generation, $rows); @@ -109,7 +113,7 @@ class BusinessMirrorSynchronizer ?Carbon $windowEnd, int $limit, ): array { - $order = $this->orderColumns($dataset, $mode); + $order = $this->orderColumns($dataset); $where = []; $bindings = []; @@ -186,22 +190,20 @@ class BusinessMirrorSynchronizer } /** @return list */ - private function orderColumns(BusinessMirrorDataset $dataset, string $mode): array + private function orderColumns(BusinessMirrorDataset $dataset): array { - $columns = $dataset->keyColumns(); - - if ($mode === 'incremental' && $dataset->watermarkColumn() !== null) { - array_unshift($columns, $dataset->watermarkColumn()); - } - - return array_values(array_unique($columns)); + // The watermark belongs in the incremental WHERE window only. Sorting + // a very large SQL Server view by an unindexed timestamp forces an + // expensive full-window sort on every keyset page and eventually times + // out. Dataset keys are the stable, unique pagination contract. + return array_values(array_unique($dataset->keyColumns())); } /** @return array */ - private function cursor(BusinessMirrorDataset $dataset, array $row, string $mode): array + private function cursor(BusinessMirrorDataset $dataset, array $row): array { $cursor = []; - foreach ($this->orderColumns($dataset, $mode) as $column) { + foreach ($this->orderColumns($dataset) as $column) { $cursor[$column] = $this->rowValue($row, $column); } diff --git a/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php index bcc8e7de..9c85136a 100644 --- a/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php @@ -40,6 +40,10 @@ class BusinessMirrorSynchronizerTest extends TestCase ], ], 'upstream_mirrors.table_sync.overlap_hours' => 24, + 'upstreams.sources.test' => [ + 'label' => 'Test source', + 'config' => 'test', + ], ]); DB::purge('business_mirror'); @@ -130,7 +134,8 @@ class BusinessMirrorSynchronizerTest extends TestCase $this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count()); $this->assertSame('新值', DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('displayName')); $this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[0]['sql']); - $this->assertStringContainsString('ORDER BY [updateTime], [Id]', $this->query->calls[0]['sql']); + $this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[0]['sql']); + $this->assertStringNotContainsString('ORDER BY [updateTime]', $this->query->calls[0]['sql']); $this->assertCount(2, $this->query->calls[0]['bindings']); foreach ($this->query->calls[0]['bindings'] as $binding) { $this->assertMatchesRegularExpression('/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}$/', $binding); @@ -138,6 +143,31 @@ class BusinessMirrorSynchronizerTest extends TestCase } } + public function test_incremental_keyset_keeps_watermark_out_of_pagination_cursor(): void + { + $this->query->batches = [[ + ['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '初始'], + ]]; + $this->synchronizer->sync('test.dbo.records', true, 10, 1); + + $this->query->calls = []; + $this->query->batches = [[ + ['Id' => 1, 'updateTime' => '2026-08-14 12:00:00', 'displayName' => '更新一'], + ['Id' => 2, 'updateTime' => '2026-08-14 12:30:00', 'displayName' => '更新二'], + ], []]; + + $result = $this->synchronizer->sync('test.dbo.records', false, 2, 0); + + $this->assertSame('succeeded', $result['status']); + $this->assertCount(2, $this->query->calls); + $this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[1]['sql']); + $this->assertStringContainsString('[Id] > ?', $this->query->calls[1]['sql']); + $this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[1]['sql']); + $this->assertStringNotContainsString('[updateTime] > ?', $this->query->calls[1]['sql']); + $this->assertCount(3, $this->query->calls[1]['bindings']); + $this->assertSame(2, $this->query->calls[1]['bindings'][2]); + } + public function test_incremental_sync_refuses_to_guess_before_first_full_sync(): void { $result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);