From 31f6c72bf3ff3bc423d1355884bc2377d94c28bd Mon Sep 17 00:00:00 2001 From: WuYongJie <549727955@qq.com> Date: Mon, 19 Oct 2026 21:04:42 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E8=AE=A2=E5=8D=95=E4=B8=8E?= =?UTF-8?q?=E8=B4=A6=E5=8D=95=E9=95=9C=E5=83=8F=E5=88=86=E6=89=B9=E8=AF=BB?= =?UTF-8?q?=E5=8F=96=E5=8F=8A=E5=86=85=E5=AD=98=E9=87=8A=E6=94=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/upstream_mirrors.php | 7 ++ .../Services/BusinessMirrorSynchronizer.php | 70 ++++++++++-- .../BusinessMirrorSynchronizerTest.php | 106 +++++++++++++++++- 3 files changed, 171 insertions(+), 12 deletions(-) diff --git a/config/upstream_mirrors.php b/config/upstream_mirrors.php index cce072c6..31d1d79d 100644 --- a/config/upstream_mirrors.php +++ b/config/upstream_mirrors.php @@ -193,6 +193,13 @@ return [ */ 'table_sync' => [ 'batch_size' => (int) env('BUSINESS_MIRROR_TABLE_BATCH_SIZE', 500), + // These wide SQL Server views time out when TOP/pagination is planned + // together with every payload column. Select the bounded keys first, + // then look up only those rows in the same read-only statement. + 'key_first_datasets' => [ + 'operations_replica.rpt.jst_Orders', + 'operations_replica.rpt.jst_originalBillInfo', + ], 'max_batches_per_run' => (int) env('BUSINESS_MIRROR_TABLE_MAX_BATCHES', 2), 'max_datasets_per_run' => (int) env('BUSINESS_MIRROR_MAX_DATASETS', 1), 'overlap_hours' => (int) env('BUSINESS_MIRROR_OVERLAP_HOURS', 48), diff --git a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php index 4c06be98..ffb83ad1 100644 --- a/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php +++ b/modules/EcommerceFinance/Services/BusinessMirrorSynchronizer.php @@ -3,9 +3,11 @@ namespace Modules\EcommerceFinance\Services; use Illuminate\Support\Carbon; +use Illuminate\Support\Facades\Log; use Illuminate\Support\Str; use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset; use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository; +use PDOException; use RuntimeException; use Throwable; @@ -64,9 +66,12 @@ class BusinessMirrorSynchronizer $batches = 0; $deadline = microtime(true) + max((int) config('upstream_pipeline.sync.dataset_seconds', 8), 1); $rows = (int) ($state['last_upserted_rows'] ?? 0); + $started = microtime(true); + $stage = 'read_upstream'; try { while (true) { + $stage = 'read_upstream'; $batch = $this->readBatch( $dataset, $columns, @@ -79,17 +84,24 @@ class BusinessMirrorSynchronizer ); if ($batch === []) { + $stage = 'complete_mirror'; $this->mirror->complete($dataset, $mode, $generation, $cursor ?? $highWatermark, $rows); return $this->result($dataset, 'succeeded', $mode, $batches, $rows); } + $stage = 'normalize_rows'; $cursor = $this->cursor($dataset, $batch[array_key_last($batch)]); - $rows += count($batch); + $batchRows = count($batch); + $rows += $batchRows; $batches++; - $this->mirror->storeBatch($dataset, $columns, $this->normalize($columns, $batch), $cursor, $generation, $rows); + $batch = $this->normalize($columns, $batch); + $stage = 'write_mirror'; + $this->mirror->storeBatch($dataset, $columns, $batch, $cursor, $generation, $rows); + unset($batch); - if (count($batch) < $batchSize) { + if ($batchRows < $batchSize) { + $stage = 'complete_mirror'; $this->mirror->complete($dataset, $mode, $generation, $cursor, $rows); return $this->result($dataset, 'succeeded', $mode, $batches, $rows); @@ -100,6 +112,28 @@ class BusinessMirrorSynchronizer } } } catch (Throwable $exception) { + // PDOException alone cannot distinguish an upstream timeout from a + // local transaction failure. Keep the stage and numeric diagnostics, + // never the exception message, SQL, bindings or connection details. + $sqlState = null; + $driverCode = null; + for ($cause = $exception; $cause !== null; $cause = $cause->getPrevious()) { + if ($cause instanceof PDOException) { + $code = (string) ($cause->errorInfo[0] ?? $cause->getCode()); + $sqlState = preg_match('/^[A-Z0-9]{5}$/D', $code) === 1 ? $code : null; + $native = $cause->errorInfo[1] ?? null; + $driverCode = is_numeric($native) ? (int) $native : null; + } + } + Log::warning('共享业务镜像同步阶段失败。', [ + 'dataset' => $dataset->key(), + 'source' => $dataset->sourceKey(), + 'stage' => $stage, + 'sql_state' => $sqlState, + 'driver_code' => $driverCode, + 'elapsed_seconds' => round(microtime(true) - $started, 3), + 'exception' => $exception::class, + ]); $this->mirror->fail($dataset, $exception::class); throw $exception; } @@ -157,33 +191,49 @@ class BusinessMirrorSynchronizer implode(', ', array_map(fn (array $column): string => $this->selectExpression($column), $columns)), $tail, $limit, ); + if ($this->driver === 'sqlsrv' && $mode === 'incremental' + && in_array($dataset->key(), (array) config('upstream_mirrors.table_sync.key_first_datasets', []), true)) { + $keys = (new UpstreamSqlDialect($this->driver))->select( + implode(', ', array_map(fn (string $name): string => $this->quote($name), $order)), $tail, $limit, + ); + $keyColumn = fn (string $name): string => $this->quote('mirror_keys') . '.' . $this->quote($name); + $rowColumn = fn (string $name): string => $this->quote('mirror_rows') . '.' . $this->quote($name); + $sql = 'SELECT ' . implode(', ', array_map(fn (array $column): string => $this->selectExpression($column, 'mirror_rows'), $columns)) + . ' FROM (' . $keys . ') AS ' . $this->quote('mirror_keys') + . ' INNER LOOP JOIN ' . $this->quote($dataset->sourceSchema()) . '.' . $this->quote($dataset->sourceObject()) + . ' AS ' . $this->quote('mirror_rows') + . ' ON ' . implode(' AND ', array_map(fn (string $name): string => $rowColumn($name) . ' = ' . $keyColumn($name), $order)) + . ' ORDER BY ' . implode(', ', array_map($keyColumn, $order)); + } + return $this->queries->forSource($dataset->sourceKey())->select($sql, $bindings); } /** @param array $column */ - private function selectExpression(array $column): string + private function selectExpression(array $column, ?string $alias = null): string { $name = (string) $column['column_name']; $quoted = $this->quote($name); + $reference = $alias === null ? $quoted : $this->quote($alias) . '.' . $quoted; if ($this->driver !== 'sqlsrv') { if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'timestamp with time zone') { // Store timezone-aware PostgreSQL instants as UTC in the shared mirror. - return "to_char({$quoted} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}"; + return "to_char({$reference} AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US') AS {$quoted}"; } if ($this->driver === 'pgsql' && ($column['data_type'] ?? '') === 'bytea') { - return "encode({$quoted}, 'hex') AS {$quoted}"; + return "encode({$reference}, 'hex') AS {$quoted}"; } - return $quoted; + return $reference; } // FreeTDS may return SQL Server date values using the connection locale // (for example "Jan 1 2026"). Convert temporal columns at the source so // every mirror receives deterministic ISO values that MariaDB accepts. return match (strtolower((string) ($column['data_type'] ?? ''))) { - 'date' => 'CONVERT(varchar(10), ' . $quoted . ', 23) AS ' . $quoted, - 'datetime', 'datetime2', 'smalldatetime' => 'CONVERT(varchar(27), ' . $quoted . ', 126) AS ' . $quoted, - default => $quoted, + 'date' => 'CONVERT(varchar(10), ' . $reference . ', 23) AS ' . $quoted, + 'datetime', 'datetime2', 'smalldatetime' => 'CONVERT(varchar(27), ' . $reference . ', 126) AS ' . $quoted, + default => $reference, }; } diff --git a/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php b/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php index 64c883b3..aae25c6f 100644 --- a/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/BusinessMirrorSynchronizerTest.php @@ -4,6 +4,7 @@ namespace Tests\Unit\EcommerceFinance; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Log; use Illuminate\Support\Facades\Schema; use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository; use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry; @@ -145,6 +146,37 @@ class BusinessMirrorSynchronizerTest extends TestCase } } + public function test_key_first_incremental_read_preserves_filters_projection_and_resume_keys(): void + { + $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '旧值']]]; + $this->synchronizer->sync('test.dbo.records', true, 10, 1); + config(['upstream_mirrors.table_sync.key_first_datasets' => ['test.dbo.records']]); + $this->query->calls = []; + $this->query->batches = [ + [['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '更新']], + [['Id' => 2, 'updateTime' => '2026-08-15 10:01:00', 'displayName' => '新增']], + [], + ]; + + $this->assertSame('running', $this->synchronizer->sync('test.dbo.records', false, 1, 1)['status']); + $this->assertSame('succeeded', $this->synchronizer->sync('test.dbo.records', false, 1, 3)['status']); + $sql = $this->query->calls[0]['sql']; + $this->assertStringContainsString('FROM (SELECT TOP (1) [Id] FROM [dbo].[records] WHERE ([updateTime] >= ? AND [updateTime] < ?) ORDER BY [Id]) AS [mirror_keys]', $sql); + $this->assertStringContainsString('INNER LOOP JOIN [dbo].[records] AS [mirror_rows] ON [mirror_rows].[Id] = [mirror_keys].[Id]', $sql); + $this->assertStringContainsString('CONVERT(varchar(27), [mirror_rows].[updateTime], 126) AS [updateTime]', $sql); + $this->assertStringContainsString('ORDER BY [mirror_keys].[Id]', $sql); + $this->assertCount(2, $this->query->calls[0]['bindings']); + $this->assertSame(1, $this->query->calls[1]['bindings'][2]); + UpstreamReadOnlyQuery::assertReadOnlySql($sql); + $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->query->calls = []; + $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '更新']]]; + $this->synchronizer->sync('test.dbo.records', true, 10, 1); + $this->assertStringNotContainsString('LOOP JOIN', $this->query->calls[0]['sql']); + } + public function test_fixed_filters_are_validated_and_bound_to_every_source_batch(): void { DB::table('upstream_schema_columns')->insert($this->column('channel', 5, 'nvarchar', 50)); @@ -507,6 +539,71 @@ class BusinessMirrorSynchronizerTest extends TestCase ); } + public function test_upstream_timeout_keeps_previous_data_and_logs_only_safe_diagnostics(): void + { + $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留']]]; + $this->synchronizer->sync('test.dbo.records', true, 10, 1); + $previousSuccess = DB::connection('business_mirror')->table('business_mirror_table_states')->value('last_succeeded_at'); + $exception = new \PDOException('secret DSN and financial SQL must never be logged'); + $exception->errorInfo = ['HY000', 20003, 'secret native message']; + $this->query->batches = [$exception]; + Log::spy(); + + try { + $this->synchronizer->sync('test.dbo.records', false, 10, 1); + $this->fail('Expected the upstream failure to propagate.'); + } catch (\PDOException $caught) { + $this->assertSame($exception, $caught); + } + + Log::shouldHaveReceived('warning')->once()->withArgs(function (string $message, array $context): bool { + $this->assertSame('read_upstream', $context['stage']); + $this->assertSame('HY000', $context['sql_state']); + $this->assertSame(20003, $context['driver_code']); + $this->assertStringNotContainsString('secret', json_encode($context)); + $this->assertArrayNotHasKey('message', $context); + $this->assertArrayNotHasKey('sql', $context); + return true; + }); + $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first(); + $this->assertSame('failed', $state->status); + $this->assertSame($previousSuccess, $state->last_succeeded_at); + $this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName')); + } + + public function test_local_pdo_failure_is_identified_as_a_mirror_write(): void + { + $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留']]]; + $this->synchronizer->sync('test.dbo.records', true, 10, 1); + $exception = new \PDOException('private local database error'); + $exception->errorInfo = ['40001', 1213, 'private message']; + $mirror = \Mockery::mock(GenericBusinessMirrorRepository::class, [app('db')])->makePartial(); + $mirror->shouldReceive('storeBatch')->once()->andThrow($exception); + $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory { + public function __construct(private readonly UpstreamReadOnlyQuery $query) {} + public function forSource(string $source): UpstreamReadOnlyQuery { return $this->query; } + }; + $synchronizer = new BusinessMirrorSynchronizer(new BusinessMirrorDatasetRegistry(), new BusinessMirrorSchemaManager(), $mirror, $factory); + $this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-15 10:00:00', 'displayName' => '未写入']]]; + Log::spy(); + + try { + $synchronizer->sync('test.dbo.records', false, 10, 1); + $this->fail('Expected the local failure to propagate.'); + } catch (\PDOException $caught) { + $this->assertSame($exception, $caught); + } + + Log::shouldHaveReceived('warning')->once()->withArgs(function (string $message, array $context): bool { + $this->assertSame('write_mirror', $context['stage']); + $this->assertSame('40001', $context['sql_state']); + $this->assertSame(1213, $context['driver_code']); + $this->assertStringNotContainsString('private', json_encode($context)); + return true; + }); + $this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName')); + } + public function test_project_dataset_registry_has_unique_safe_targets(): void { $configured = require base_path('config/upstream_mirrors.php'); @@ -561,7 +658,7 @@ class BusinessMirrorSynchronizerTest extends TestCase class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery { - /** @var list>> */ + /** @var list>|\Throwable> */ public array $batches = []; /** @var list */ @@ -576,6 +673,11 @@ class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery { $this->calls[] = compact('sql', 'bindings'); - return array_shift($this->batches) ?? []; + $batch = array_shift($this->batches) ?? []; + if ($batch instanceof \Throwable) { + throw $batch; + } + + return $batch; } }