diff --git a/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php b/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php index c533b960..90b7f617 100644 --- a/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php +++ b/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php @@ -25,6 +25,8 @@ use Throwable; */ class ShippingReplicaSynchronizer { + private const UPSTREAM_RECONNECT_RETRIES = 2; + private const SOURCE = 'operations_replica'; private const DATASET = 'jst.sctrade'; private const FREIGHT_DATASET_PREFIX = 'jst.sctrade.freight.'; @@ -383,17 +385,18 @@ class ShippingReplicaSynchronizer /** @return list> */ private function selectWithReconnect(string $sql, array $bindings): array { - try { - return $this->up->select($sql, $bindings); - } catch (PDOException $e) { - if (! $this->isTransientCursorFailure($e)) { - throw $e; + for ($retry = 0; ; $retry++) { + try { + return $this->up->select($sql, $bindings); + } catch (PDOException $e) { + if (! $this->isTransientCursorFailure($e) || $retry >= self::UPSTREAM_RECONNECT_RETRIES) { + throw $e; + } + + // select() returns only after fetchAll(), so no partial result was consumed. + UpstreamConnection::reset(self::SOURCE); + usleep(250000); } - - // select() returns only after fetchAll(), so no partial result was consumed. - UpstreamConnection::reset(self::SOURCE); - - return $this->up->select($sql, $bindings); } } @@ -401,7 +404,8 @@ class ShippingReplicaSynchronizer { $message = strtolower($e->getMessage()); - return str_contains($message, 'pdo_dblib: dbnextrow() returned fail'); + return str_contains($message, 'pdo_dblib: dbnextrow() returned fail') + || str_contains($message, 'adaptive server connection timed out [20003]'); } /** 并入自动发货仓:jst_Orders 已发货但不在发货明细表。 */ diff --git a/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php b/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php index 3011fd7b..b3329b68 100644 --- a/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php @@ -178,6 +178,39 @@ class ShippingReplicaSynchronizerTest extends TestCase ->where('dataset_key', 'jst.sctrade')->value('status')); } + public function test_adaptive_server_timeout_reconnects_and_replays_the_same_page(): void + { + $page = array_map(fn (int $id): array => [ + 'id' => $id, 'send_time' => $id === 1500 ? '2026-09-02 12:00:00' : null, + ], range(1500, 1)); + $scanCalls = 0; + $bindings = []; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql, array $bind) use (&$scanCalls, &$bindings, $page): array { + if (! str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + return []; + } + $scanCalls++; + $bindings[] = $bind; + if ($scanCalls <= 2) { + throw new PDOException('SQLSTATE[HY000]: General error: 20003 Adaptive Server connection timed out [20003] (severity 6)'); + } + return $scanCalls === 3 ? $page : []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + $result = (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true, + ); + + $this->assertSame(4, $scanCalls); + $this->assertSame([[], [], [], [1]], $bindings); + $this->assertSame(1, $result['processed']); + $this->assertSame([1500], DB::connection('business_mirror')->table('jst_oms_shipping') + ->pluck('source_id')->map(fn ($id) => (int) $id)->all()); + } + public function test_window_result_distinguishes_duplicate_package_rows_from_unique_source_ids(): void { $upstream = $this->createMock(UpstreamReadOnlyQuery::class);