WuYongJie/CaiWuXiTong
1
0
Fork 0
Code Issues Pull Requests 5 Packages Projects Releases Wiki Activity

fix: avoid shipping sync replica timeout

Browse Source
This commit is contained in:
limeng 2026-09-24 09:56:58 +08:00
parent a5a17a5c13
commit bee27aa2be
2 changed files with 53 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
modules/EcommerceFinance/Services/Jushuitan/ShippingReplicaSynchronizer.php
View File

@ -200,7 +200,7 @@ class ShippingReplicaSynchronizer
$logs[$lg] = true;
}
}
[$byOid, $bySo] = $this->loadAllItems(array_keys($oids), array_keys($soIds));
[$byOid, $bySo] = $this->loadAllItems(array_keys($oids), array_keys($soIds), ! $windowed);
$dws = [];
foreach (array_keys($soIds) as $mso) {
if (empty($bySo[$mso]) && ! empty($this->dwsCache[$mso])) {
@ -311,7 +311,7 @@ class ShippingReplicaSynchronizer
}
$lastRid = $r['rid'];
}
[$byOid, $bySo] = $this->loadAllItems(array_keys($oids), array_keys($soIds));
[$byOid, $bySo] = $this->loadAllItems(array_keys($oids), array_keys($soIds), false);
$dws = [];
foreach (array_keys($soIds) as $mso) {
if (empty($bySo[$mso]) && ! empty($this->dwsCache[$mso])) {
@ -518,7 +518,7 @@ class ShippingReplicaSynchronizer
}
/** @return array{0:array<string,array>,1:array<string,array>} [byOid, bySo] */
private function loadAllItems(array $oids, array $soIds): array
private function loadAllItems(array $oids, array $soIds, bool $loadBySoId = true): array
{
$byOid = [];
$bySo = [];
@ -536,15 +536,21 @@ class ShippingReplicaSynchronizer
}
}
}
foreach (array_chunk($soIds, 800) as $chunk) {
if ($chunk === []) {
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) {
$k = (string) ($r['raw_so_id'] ?? '');
if ($k !== '') {
$bySo[$k][] = $this->itemArr($r);
// raw_so_id lacks a usable index in the production replica. Windowed sync already
// preloads DWS rows by sourceno, so querying this fallback can turn every batch into
// a full-table scan and exceed the upstream timeout. Keep it only for the small,
// non-windowed diagnostic mode where DWS is not preloaded.
if ($loadBySoId) {
foreach (array_chunk($soIds, 800) as $chunk) {
if ($chunk === []) {
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) {
$k = (string) ($r['raw_so_id'] ?? '');
if ($k !== '') {
$bySo[$k][] = $this->itemArr($r);
}
}
}
}

35
tests/Unit/EcommerceFinance/ShippingReplicaSynchronizerTest.php
View File

@ -10,6 +10,41 @@ use Tests\TestCase;
class ShippingReplicaSynchronizerTest extends TestCase
{
public function test_windowed_item_lookup_skips_unindexed_raw_so_id_fallback(): void
{
$upstream = $this->createMock(UpstreamReadOnlyQuery::class);
$upstream->expects($this->once())
->method('select')
->with(
$this->stringContains('WHERE o_id IN'),
['order-1'],
)
->willReturn([
[
'o_id' => 'order-1',
'raw_so_id' => 'trade-1',
'sku_id' => 'sku-1',
'i_id' => 'item-1',
'shop_i_id' => 'shop-item-1',
'name' => '商品',
'qty' => 1,
'amount' => 10,
'send_warehouse' => '仓库',
'Id' => 1,
],
]);
$factory = $this->createMock(UpstreamReadOnlyQueryFactory::class);
$factory->method('forSource')->with('operations_replica')->willReturn($upstream);
$synchronizer = new ShippingReplicaSynchronizer($factory);
$method = new \ReflectionMethod($synchronizer, 'loadAllItems');
$method->setAccessible(true);
[$byOrder, $byTrade] = $method->invoke($synchronizer, ['order-1'], ['trade-1'], false);
$this->assertArrayHasKey('order-1', $byOrder);
$this->assertSame([], $byTrade);
}
public function test_sync_removes_only_the_known_demo_source_ids(): void
{
$db = DB::connection('business_mirror');