@@ -658,31 +658,36 @@ export class TelegramPollingSession {
|
658 | 658 | return deferredSpooledUpdateClaimsByKey.has(buildDeferredSpooledUpdateClaimKey(update)); |
659 | 659 | } |
660 | 660 | |
661 | | - #isTimedOutSpooledUpdateClaim(update: ClaimedTelegramSpooledUpdate): boolean { |
662 | | -const claimedAt = update.claim?.claimedAt; |
663 | | -return claimedAt !== undefined && Date.now() - claimedAt >= this.#spooledUpdateHandlerTimeoutMs; |
664 | | -} |
665 | | - |
666 | | -async #failTimedOutLiveOwnedSpooledUpdateClaims(params: { |
| 661 | +async #failTimedOutCurrentProcessSpooledUpdateClaims(params: { |
667 | 662 | activeLaneKeys: Set<string>; |
668 | 663 | spoolDir: string; |
669 | 664 | }): Promise<void> { |
670 | 665 | const claims = await listTelegramSpooledUpdateClaims({ spoolDir: params.spoolDir }); |
| 666 | +const now = Date.now(); |
671 | 667 | for (const claim of claims) { |
| 668 | +const claimOwner = claim.claim; |
| 669 | +if (!claimOwner) { |
| 670 | +continue; |
| 671 | +} |
672 | 672 | if (this.#isDeferredSpooledUpdateClaim(claim)) { |
673 | 673 | continue; |
674 | 674 | } |
675 | 675 | if (params.activeLaneKeys.has(this.#spooledUpdateLaneKey(claim))) { |
676 | 676 | continue; |
677 | 677 | } |
678 | | -if (!this.#isTimedOutSpooledUpdateClaim(claim)) { |
| 678 | +if (now - claimOwner.claimedAt < this.#spooledUpdateHandlerTimeoutMs) { |
679 | 679 | continue; |
680 | 680 | } |
681 | 681 | if (!isTelegramSpooledUpdateClaimOwnedByOtherLiveProcess(claim)) { |
682 | 682 | continue; |
683 | 683 | } |
684 | | -const claimedForMs = Date.now() - (claim.claim?.claimedAt ?? Date.now()); |
685 | | -const message = `Telegram spooled update claim held by a live worker for ${formatDurationPrecise(claimedForMs)} without active handler state; marking failed so the lane can continue.`; |
| 684 | +// Same PID with a stale owner id means this process orphaned a previous |
| 685 | +// local handler state; different live PIDs may still be processing. |
| 686 | +if (claimOwner.processPid !== process.pid) { |
| 687 | +continue; |
| 688 | +} |
| 689 | +const claimedForMs = now - claimOwner.claimedAt; |
| 690 | +const message = `Telegram spooled update claim owned by this process for ${formatDurationPrecise(claimedForMs)} without active handler state; marking failed so the lane can continue.`; |
686 | 691 | try { |
687 | 692 | const failed = await failTelegramSpooledUpdateClaim({ |
688 | 693 | update: claim, |
@@ -691,13 +696,13 @@ export class TelegramPollingSession {
|
691 | 696 | }); |
692 | 697 | if (!failed) { |
693 | 698 | this.opts.log( |
694 | | -`[telegram][diag] spooled update ${claim.updateId} live-owned claim no longer had a processing marker to fail.`, |
| 699 | +`[telegram][diag] spooled update ${claim.updateId} current-process claim no longer had a processing marker to fail.`, |
695 | 700 | ); |
696 | 701 | continue; |
697 | 702 | } |
698 | 703 | } catch (err) { |
699 | 704 | this.opts.log( |
700 | | -`[telegram][diag] spooled update ${claim.updateId} live-owned claim could not be marked failed: ${formatErrorMessage(err)}`, |
| 705 | +`[telegram][diag] spooled update ${claim.updateId} current-process claim could not be marked failed: ${formatErrorMessage(err)}`, |
701 | 706 | ); |
702 | 707 | continue; |
703 | 708 | } |
@@ -828,7 +833,7 @@ export class TelegramPollingSession {
|
828 | 833 | spoolDir: string; |
829 | 834 | }): Promise<SpooledUpdateDrainResult> { |
830 | 835 | const activeLaneKeys = this.#activeSpooledUpdateLaneKeysForSpool(params.spoolDir); |
831 | | -await this.#failTimedOutLiveOwnedSpooledUpdateClaims({ |
| 836 | +await this.#failTimedOutCurrentProcessSpooledUpdateClaims({ |
832 | 837 | activeLaneKeys, |
833 | 838 | spoolDir: params.spoolDir, |
834 | 839 | }); |
|
