重试发货同步的上游连接超时 #272
@ -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<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;
|
||||
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 已发货但不在发货明细表。 */
|
||||
|
||||
@ -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);
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user