WuYongJie/CaiWuXiTong
1
0
Fork 0
Code Issues Pull Requests 7 Packages Projects Releases Wiki Activity

修复业务镜像增量分页超时 (#61)

Browse Source
This commit is contained in:
WuYongJie 2026-08-15 05:03:19 +08:00
parent fbd05380f3
commit 809bca2e5c
2 changed files with 45 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php
View File

@ -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<string> */
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<string,mixed> */
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);
}

32
tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php
View File

@ -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);