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<array<string,mixed>> */
+    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;

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