feat(desktop-client): queue account probes and report health to server

Replace serial probe-on-lock with a bounded probe queue (concurrency 3) that adds a "queued" probe state, manual cooldown, and retry backoff. After each probe the runtime debounces a batched POST to the new /desktop/accounts/health-reports endpoint so the server learns about live, expired, and risk transitions without waiting for a re-bind. Adds an account-scoped IPC (probeRuntimeAccount + runtimeAccountSnapshot) so the renderer can refresh a single row instead of replaying the full snapshot, and exposes the resulting "重新校验" action in AccountsView/AiPlatformsView. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-27 21:33:55 +08:00
parent 051976e4a9
commit c3feb7477a
15 changed files with 817 additions and 129 deletions
@@ -30,6 +30,12 @@ const CONFIRM_BACKOFF_STEPS_MS = [30_000, 120_000] as const;
 const NETWORK_RETRY_MS = 2 * 60_000;
 const STALE_AFTER_MS = 45 * 60_000;
 const PREFLIGHT_MAX_AGE_MS = 15 * 60_000;
+const MAX_CONCURRENT_ACCOUNT_PROBES = 3;
+const MANUAL_PROBE_CACHE_TTL_MS = 5 * 60_000;
+const MANUAL_PROBE_MIN_INTERVAL_MS = 15_000;
+const ACCOUNT_PROBE_RETRY_BACKOFF_MS = [1_000, 3_000, 10_000] as const;
+
+type ProbeTrigger = "scheduled" | "manual" | "preflight" | "confirm";

 interface PersistedAccountHealthRecord {
  accountId: string;
@@ -56,11 +62,27 @@ interface AccountHealthRecord {
  confirmFailureCount: number;
 }

+interface AccountProbeJob {
+  account: PublishAccountIdentity;
+  trigger: ProbeTrigger;
+  allowSilentRefresh: boolean;
+  force: boolean;
+  status: "queued" | "running";
+  attempt: number;
+  maxAttempts: number;
+  resolve: (snapshot: AccountHealthSnapshot) => void;
+  reject: (error: unknown) => void;
+  promise: Promise<AccountHealthSnapshot>;
+}
+
 const trackedAccounts = new Map<string, PublishAccountIdentity>();
 const records = new Map<string, AccountHealthRecord>();
 const accountLocks = new Map<string, Promise<void>>();
+const probeJobs = new Map<string, AccountProbeJob>();
+const probeQueue: string[] = [];
 let initialized = false;
 let probeTimer: ReturnType<typeof setInterval> | null = null;
+let activeProbeCount = 0;

 function randomInt(min: number, max: number): number {
  if (max <= min) {
@@ -193,7 +215,7 @@ function commitRecord(record: AccountHealthRecord, previousSignature: string | n

  const nextSignature = snapshotSignature(record);
  if (previousSignature !== nextSignature) {
-    emitRuntimeInvalidated("account-health");
+    emitRuntimeInvalidated("account-health", { accountId: record.accountId });
  }

  return toPublicSnapshot(record);
@@ -342,7 +364,7 @@ function applyNetworkResult(
 async function performProbeLocked(
  account: PublishAccountIdentity,
  options: {
-    trigger: "scheduled" | "manual" | "preflight" | "confirm";
+    trigger: ProbeTrigger;
    allowSilentRefresh: boolean;
  },
 ): Promise<AccountHealthSnapshot> {
@@ -384,6 +406,165 @@ async function performProbeLocked(
  }
 }

+function canUseCachedProbe(
+  record: AccountHealthRecord,
+  options: { trigger: ProbeTrigger; force: boolean },
+  now = Date.now(),
+): boolean {
+  if (options.force) {
+    return Boolean(record.lastProbeAt && now - record.lastProbeAt < MANUAL_PROBE_MIN_INTERVAL_MS);
+  }
+
+  if (
+    options.trigger === "manual"
+    && record.lastProbeAt
+    && now - record.lastProbeAt < MANUAL_PROBE_MIN_INTERVAL_MS
+  ) {
+    return true;
+  }
+
+  if (
+    (options.trigger === "manual" || options.trigger === "scheduled")
+    && record.authState === "active"
+    && record.lastVerifiedAt
+    && now - record.lastVerifiedAt < MANUAL_PROBE_CACHE_TTL_MS
+    && record.probeState !== "network_error"
+  ) {
+    return true;
+  }
+
+  return false;
+}
+
+function markProbeQueued(record: AccountHealthRecord): AccountHealthSnapshot {
+  const previousSignature = snapshotSignature(record);
+  record.probeState = "queued";
+  return commitRecord(record, previousSignature);
+}
+
+function maxAttemptsForTrigger(trigger: ProbeTrigger): number {
+  return trigger === "preflight" ? 1 : ACCOUNT_PROBE_RETRY_BACKOFF_MS.length + 1;
+}
+
+function enqueueProbeJob(job: AccountProbeJob): void {
+  if (!probeQueue.includes(job.account.id)) {
+    probeQueue.push(job.account.id);
+  }
+  drainProbeQueue();
+}
+
+function scheduleProbeRetry(job: AccountProbeJob): void {
+  const retryIndex = Math.min(Math.max(job.attempt - 2, 0), ACCOUNT_PROBE_RETRY_BACKOFF_MS.length - 1);
+  const backoff = ACCOUNT_PROBE_RETRY_BACKOFF_MS[retryIndex]
+    ?? ACCOUNT_PROBE_RETRY_BACKOFF_MS[ACCOUNT_PROBE_RETRY_BACKOFF_MS.length - 1];
+  job.status = "queued";
+  setTimeout(() => enqueueProbeJob(job), backoff);
+}
+
+function drainProbeQueue(): void {
+  while (activeProbeCount < MAX_CONCURRENT_ACCOUNT_PROBES && probeQueue.length > 0) {
+    const accountId = probeQueue.shift();
+    if (!accountId) {
+      continue;
+    }
+
+    const job = probeJobs.get(accountId);
+    if (!job || job.status !== "queued") {
+      continue;
+    }
+
+    job.status = "running";
+    activeProbeCount += 1;
+    void runProbeJob(job);
+  }
+}
+
+async function runProbeJob(job: AccountProbeJob): Promise<void> {
+  try {
+    const snapshot = await withAccountLock(job.account.id, async () => {
+      const record = ensureRecord(job.account);
+      if (canUseCachedProbe(record, job)) {
+        return toPublicSnapshot(record);
+      }
+      return await performProbeLocked(job.account, {
+        trigger: job.trigger,
+        allowSilentRefresh: job.allowSilentRefresh,
+      });
+    });
+
+    if (snapshot.probeState === "network_error" && job.attempt < job.maxAttempts) {
+      job.attempt += 1;
+      scheduleProbeRetry(job);
+      return;
+    }
+
+    probeJobs.delete(job.account.id);
+    job.resolve(snapshot);
+  } catch (error) {
+    probeJobs.delete(job.account.id);
+    job.reject(error);
+  } finally {
+    activeProbeCount = Math.max(0, activeProbeCount - 1);
+    drainProbeQueue();
+  }
+}
+
+function enqueueAccountProbe(
+  account: PublishAccountIdentity,
+  options: {
+    trigger?: ProbeTrigger;
+    allowSilentRefresh?: boolean;
+    force?: boolean;
+  } = {},
+): Promise<AccountHealthSnapshot> {
+  syncTrackedAccountsInternal([account], false);
+
+  const trigger = options.trigger ?? "manual";
+  const force = Boolean(options.force);
+  const record = ensureRecord(account);
+  if (canUseCachedProbe(record, { trigger, force })) {
+    return Promise.resolve(toPublicSnapshot(record));
+  }
+
+  const existing = probeJobs.get(account.id);
+  if (existing) {
+    if (force) {
+      existing.force = true;
+    }
+    existing.allowSilentRefresh = existing.allowSilentRefresh || Boolean(options.allowSilentRefresh);
+    if (trigger === "manual") {
+      existing.trigger = "manual";
+    }
+    return existing.promise;
+  }
+
+  markProbeQueued(record);
+
+  let resolve!: (snapshot: AccountHealthSnapshot) => void;
+  let reject!: (error: unknown) => void;
+  const promise = new Promise<AccountHealthSnapshot>((resolvePromise, rejectPromise) => {
+    resolve = resolvePromise;
+    reject = rejectPromise;
+  });
+
+  const job: AccountProbeJob = {
+    account,
+    trigger,
+    allowSilentRefresh: Boolean(options.allowSilentRefresh),
+    force,
+    status: "queued",
+    attempt: 1,
+    maxAttempts: maxAttemptsForTrigger(trigger),
+    resolve,
+    reject,
+    promise,
+  };
+
+  probeJobs.set(account.id, job);
+  enqueueProbeJob(job);
+  return promise;
+}
+
 async function runDueProbes(): Promise<void> {
  const now = Date.now();
  const dueAccounts = [...trackedAccounts.values()].filter((account) => {
@@ -398,10 +579,14 @@ async function runDueProbes(): Promise<void> {
    }

    const trigger = record.suspectedExpiredAt ? "confirm" : "scheduled";
-    void withAccountLock(account.id, async () => {
-      await performProbeLocked(account, {
-        trigger,
-        allowSilentRefresh: false,
+    void enqueueAccountProbe(account, {
+      trigger,
+      allowSilentRefresh: false,
+    }).catch((error) => {
+      console.warn("[desktop-account-health] scheduled probe failed", {
+        accountId: account.id,
+        platform: account.platform,
+        message: error instanceof Error ? error.message : String(error),
      });
    });
  }
@@ -454,6 +639,7 @@ function syncTrackedAccountsInternal(
        trackedAccounts.delete(accountId);
        records.delete(accountId);
        removed = true;
+        emitRuntimeInvalidated("account-health", { accountId });
      }
    }
  }
@@ -475,7 +661,7 @@ export function forgetTrackedAccountHealth(
  trackedAccounts.delete(accountId);
  if (records.delete(accountId)) {
    persistRecords();
-    emitRuntimeInvalidated("account-health");
+    emitRuntimeInvalidated("account-health", { accountId });
  }
 }

@@ -526,16 +712,15 @@ export function getProjectedAccountHealth(input: {
 export async function probeTrackedAccount(
  account: PublishAccountIdentity,
  options: {
-    trigger?: "scheduled" | "manual" | "preflight" | "confirm";
+    trigger?: ProbeTrigger;
    allowSilentRefresh?: boolean;
+    force?: boolean;
  } = {},
 ): Promise<AccountHealthSnapshot> {
-  syncTrackedAccountsInternal([account], false);
-  return await withAccountLock(account.id, async () => {
-    return await performProbeLocked(account, {
-      trigger: options.trigger ?? "manual",
-      allowSilentRefresh: Boolean(options.allowSilentRefresh),
-    });
+  return await enqueueAccountProbe(account, {
+    trigger: options.trigger ?? "manual",
+    allowSilentRefresh: Boolean(options.allowSilentRefresh),
+    force: Boolean(options.force),
  });
 }