diff --git a/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php b/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php index 304cd253..c533b960 100644 --- a/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php +++ b/modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php @@ -5,7 +5,9 @@ namespace Modules\EcommerceFinance\Services\Jushuitan; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\DB; use InvalidArgumentException; +use Modules\EcommerceFinance\Services\UpstreamConnection; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory; +use PDOException; use RuntimeException; use Throwable; @@ -265,7 +267,7 @@ class ShippingReplicaSynchronizer $sql = "SELECT TOP {$batch} {$cols} FROM rpt.jst_order_express_details e " . 'LEFT JOIN rpt.jst_Orders jo ON jo.o_id = e.inner_order_id ' . "WHERE {$where} ORDER BY e.id DESC"; - $rows = $this->up->select($sql, $bind); + $rows = $this->selectWithReconnect($sql, $bind); if ($rows === []) { break; } @@ -378,6 +380,30 @@ class ShippingReplicaSynchronizer return $total; } + /** @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; + } + + // select() returns only after fetchAll(), so no partial result was consumed. + UpstreamConnection::reset(self::SOURCE); + + return $this->up->select($sql, $bindings); + } + } + + private function isTransientCursorFailure(PDOException $e): bool + { + $message = strtolower($e->getMessage()); + + return str_contains($message, 'pdo_dblib: dbnextrow() returned fail'); + } + /** 并入自动发货仓:jst_Orders 已发货但不在发货明细表。 */ private function syncAutoShip(string $fromS, string $toS, $now, ?callable $progress = null): int { @@ -399,7 +425,7 @@ class ShippingReplicaSynchronizer $bind[] = $lastRid; } $sql = "SELECT TOP {$batch} {$cols} FROM rpt.jst_Orders o WHERE {$where} ORDER BY o.Id DESC"; - $rows = $this->up->select($sql, $bind); + $rows = $this->selectWithReconnect($sql, $bind); if ($rows === []) { break; } @@ -666,7 +692,7 @@ class ShippingReplicaSynchronizer continue; } $ph = implode(',', array_fill(0, count($chunk), '?')); - foreach ($this->up->select("SELECT {$sel} FROM rpt.jst_OrderItem WHERE o_id IN ({$ph}) ORDER BY Id", $chunk) as $r) { + foreach ($this->selectWithReconnect("SELECT {$sel} FROM rpt.jst_OrderItem WHERE o_id IN ({$ph}) ORDER BY Id", $chunk) as $r) { $k = (string) ($r['o_id'] ?? ''); if ($k !== '') { $byOid[$k][] = $this->itemArr($r); @@ -683,7 +709,7 @@ class ShippingReplicaSynchronizer continue; } $ph = implode(',', array_fill(0, count($chunk), '?')); - foreach ($this->up->select("SELECT {$sel} FROM rpt.jst_OrderItem WHERE raw_so_id IN ({$ph}) ORDER BY Id", $chunk) as $r) { + foreach ($this->selectWithReconnect("SELECT {$sel} FROM rpt.jst_OrderItem WHERE raw_so_id IN ({$ph}) ORDER BY Id", $chunk) as $r) { $k = (string) ($r['raw_so_id'] ?? ''); if ($k !== '') { $bySo[$k][] = $this->itemArr($r); @@ -706,7 +732,7 @@ class ShippingReplicaSynchronizer $d = $day->format('Y-m-d'); $day = $day->addDay(); try { - $rows = $this->up->select("SELECT sourceno, item_id, skuoid, goods_count, sales_amount FROM rpt.dws_sn_sales_full_info_di WHERE stat_date = ?", [$d]); + $rows = $this->selectWithReconnect("SELECT sourceno, item_id, skuoid, goods_count, sales_amount FROM rpt.dws_sn_sales_full_info_di WHERE stat_date = ?", [$d]); } catch (Throwable) { $this->dwsFailedDays[] = $d; continue; @@ -742,7 +768,7 @@ class ShippingReplicaSynchronizer } $ph = implode(',', array_fill(0, count($chunk), '?')); $sql = "SELECT sourceno, item_id, skuoid, goods_count, sales_amount FROM rpt.dws_sn_sales_full_info_di WHERE sourceno IN ({$ph})"; - foreach ($this->up->select($sql, $chunk) as $r) { + foreach ($this->selectWithReconnect($sql, $chunk) as $r) { $so = (string) ($r['sourceno'] ?? ''); $barcode = (string) ($r['skuoid'] ?? ''); if ($so === '' || $barcode === '') { @@ -793,7 +819,7 @@ class ShippingReplicaSynchronizer { $this->outboundMap = ['byL' => [], 'byO' => []]; $sql = 'SELECT l_id, o_id, io_id FROM ' . self::OUTBOUND . ' WHERE io_date >= ? AND io_date < ?'; - foreach ($this->up->select($sql, [$fromS, $toS]) as $r) { + foreach ($this->selectWithReconnect($sql, [$fromS, $toS]) as $r) { $io = (string) ($r['io_id'] ?? ''); if ($io === '') { continue; @@ -824,7 +850,7 @@ class ShippingReplicaSynchronizer continue; } $ph = implode(',', array_fill(0, count($chunk), '?')); - foreach ($this->up->select('SELECT l_id, io_id FROM ' . self::OUTBOUND . " WHERE l_id IN ({$ph})", $chunk) as $r) { + foreach ($this->selectWithReconnect('SELECT l_id, io_id FROM ' . self::OUTBOUND . " WHERE l_id IN ({$ph})", $chunk) as $r) { $l = (string) ($r['l_id'] ?? ''); if ($l !== '' && ! isset($byL[$l])) { $byL[$l] = (string) ($r['io_id'] ?? ''); @@ -836,7 +862,7 @@ class ShippingReplicaSynchronizer continue; } $ph = implode(',', array_fill(0, count($chunk), '?')); - foreach ($this->up->select('SELECT o_id, io_id FROM ' . self::OUTBOUND . " WHERE o_id IN ({$ph})", $chunk) as $r) { + foreach ($this->selectWithReconnect('SELECT o_id, io_id FROM ' . self::OUTBOUND . " WHERE o_id IN ({$ph})", $chunk) as $r) { $o = (string) ($r['o_id'] ?? ''); if ($o !== '' && ! isset($byO[$o])) { $byO[$o] = (string) ($r['io_id'] ?? ''); @@ -870,7 +896,7 @@ class ShippingReplicaSynchronizer $ph = implode(',', array_fill(0, count($chunk), '?')); $sql = 'SELECT io_id, sku_ids, qty, type, sku_name, shop_i_id, wareHoursName, l, w, h, volume, weight, supplier_name FROM ' . self::VIEW . " WHERE io_id IN ({$ph})"; - foreach ($this->up->select($sql, $chunk) as $r) { + foreach ($this->selectWithReconnect($sql, $chunk) as $r) { $io = (string) ($r['io_id'] ?? ''); $skuIds = (string) ($r['sku_ids'] ?? ''); if ($io === '' || $skuIds === '') { @@ -1196,7 +1222,7 @@ class ShippingReplicaSynchronizer continue; } $ph = implode(',', array_fill(0, count($chunk), '?')); - foreach ($this->up->select("SELECT sku_id, l, w, h, pack_volume, weight, supplier_name, name, i_id FROM rpt.jst_sku WHERE sku_id IN ({$ph})", $chunk) as $r) { + foreach ($this->selectWithReconnect("SELECT sku_id, l, w, h, pack_volume, weight, supplier_name, name, i_id FROM rpt.jst_sku WHERE sku_id IN ({$ph})", $chunk) as $r) { $out[(string) $r['sku_id']] = [ 'l' => $r['l'] ?? null, 'w' => $r['w'] ?? null, 'h' => $r['h'] ?? null, 'vol' => $r['pack_volume'] ?? null, 'weight' => $r['weight'] ?? null, @@ -1221,7 +1247,7 @@ class ShippingReplicaSynchronizer $sql = 'SELECT so_id, expressNo, SUM(freightCost) AS cost, SUM(weight) AS w, ' . 'MAX(expressName) AS name, MAX(receiver_state) AS state ' . "FROM rpt.fys_jstFreightCostDetail WHERE so_id IN ({$ph}) GROUP BY so_id, expressNo"; - foreach ($this->up->select($sql, $chunk) as $r) { + foreach ($this->selectWithReconnect($sql, $chunk) as $r) { $expno = (string) ($r['expressNo'] ?? ''); $val = ['cost' => $r['cost'] ?? null, 'w' => $r['w'] ?? null, 'name' => $r['name'] ?? null, 'state' => $r['state'] ?? null, 'no' => $expno ?: null]; $out[((string) $r['so_id']) . '|' . $expno] = $val; diff --git a/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php b/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php index 059cbcbe..3011fd7b 100644 --- a/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php @@ -8,6 +8,7 @@ use Modules\EcommerceFinance\Services\Jushuitan\ShippingReplicaSynchronizer; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery; use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory; use Modules\EcommerceFinance\Repositories\ShippingOrderRepository; +use PDOException; use RuntimeException; use Tests\TestCase; @@ -111,6 +112,72 @@ class ShippingReplicaSynchronizerTest extends TestCase ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all()); } + public function test_express_page_reconnects_once_after_transient_dblib_cursor_failure(): void + { + $firstPage = array_map(fn (int $id): array => [ + 'id' => $id, + 'send_time' => $id === 3000 ? '2026-09-02 12:00:00' : null, + ], range(3000, 1501)); + $lastPage = [['id' => 1500, 'send_time' => '2026-09-03 12:00:00']]; + $scanCalls = 0; + $bindings = []; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql, array $bind) use (&$scanCalls, &$bindings, $firstPage, $lastPage): 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: PDO_DBLIB: dbnextrow() returned FAIL'); + } + + return $scanCalls === 1 ? $firstPage : $lastPage; + }); + $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(3, $scanCalls); + $this->assertSame([[], [1501], [1501]], $bindings); + $this->assertSame(2, $result['processed']); + $this->assertSame(2, $result['unique_packages']); + $this->assertSame([1500, 3000], DB::connection('business_mirror')->table('jst_oms_shipping') + ->orderBy('source_id')->pluck('source_id')->map(fn ($id) => (int) $id)->all()); + } + + public function test_non_transient_upstream_failure_is_not_retried(): void + { + $calls = 0; + $upstream = $this->createMock(UpstreamReadOnlyQuery::class); + $upstream->method('select')->willReturnCallback(function (string $sql) use (&$calls): array { + if (str_contains($sql, 'FROM rpt.jst_order_express_details e LEFT JOIN')) { + $calls++; + throw new PDOException('SQLSTATE[HY000]: unrelated upstream error'); + } + + return []; + }); + $factory = $this->createMock(UpstreamReadOnlyQueryFactory::class); + $factory->method('forSource')->with('operations_replica')->willReturn($upstream); + + try { + (new ShippingReplicaSynchronizer($factory))->sync( + Carbon::parse('2026-09-01'), Carbon::parse('2026-09-05'), 2000, null, true, + ); + $this->fail('Expected the upstream error to be propagated.'); + } catch (PDOException $e) { + $this->assertStringContainsString('unrelated upstream error', $e->getMessage()); + } + + $this->assertSame(1, $calls); + $this->assertSame('failed', DB::connection('business_mirror')->table('jst_sync_state') + ->where('dataset_key', 'jst.sctrade')->value('status')); + } + public function test_window_result_distinguishes_duplicate_package_rows_from_unique_source_ids(): void { $upstream = $this->createMock(UpstreamReadOnlyQuery::class);