mirror of
https://github.com/toeverything/AFFiNE.git
synced 2026-07-29 08:09:52 +08:00
feat: handle calendar sync failed (#14510)
#### PR Dependency Tree * **PR #14510** 👈 This tree was auto-generated by [Charcoal](https://github.com/danerwilliams/charcoal) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Bug Fixes** * Improved calendar sync reliability with exponential backoff for repeated failures. * Better handling of token refresh failures with automatic account invalidation and cleanup when needed. * Subscriptions are now automatically disabled and related events removed when the calendar provider reports missing resources. * **Tests** * Added comprehensive tests covering sync failures, backoff behavior, token refresh handling, skipping retries during backoff, and recovery. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
DarkSky
@@ -12,6 +12,7 @@ import {
|
||||
Mutex,
|
||||
URLHelper,
|
||||
} from '../../base';
|
||||
import { SessionRedis } from '../../base/redis';
|
||||
import { Models } from '../../models';
|
||||
import type { CalendarCalDAVProviderPreset } from './config';
|
||||
import {
|
||||
@@ -27,6 +28,10 @@ import type { LinkCalDAVAccountInput } from './types';
|
||||
const TOKEN_REFRESH_SKEW_MS = 60 * 1000;
|
||||
const DEFAULT_PAST_DAYS = 90;
|
||||
const DEFAULT_FUTURE_DAYS = 180;
|
||||
const SYNC_FAILURE_BACKOFF_KEY_PREFIX = 'calendar:sync:backoff:';
|
||||
const SYNC_FAILURE_BACKOFF_BASE_MS = 5 * 60 * 1000;
|
||||
const SYNC_FAILURE_BACKOFF_MAX_MS = 6 * 60 * 60 * 1000;
|
||||
const SYNC_FAILURE_BACKOFF_TTL_SECONDS = 24 * 60 * 60;
|
||||
|
||||
@Injectable()
|
||||
export class CalendarService {
|
||||
@@ -37,6 +42,7 @@ export class CalendarService {
|
||||
private readonly models: Models,
|
||||
private readonly providerFactory: CalendarProviderFactory<CalendarProvider>,
|
||||
private readonly mutex: Mutex,
|
||||
private readonly redis: SessionRedis,
|
||||
private readonly config: Config,
|
||||
private readonly url: URLHelper
|
||||
) {}
|
||||
@@ -307,6 +313,12 @@ export class CalendarService {
|
||||
return;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const backoff = await this.getSyncFailureBackoff(subscription.id);
|
||||
if (backoff && now < backoff.nextRetryAt.getTime()) {
|
||||
return;
|
||||
}
|
||||
|
||||
await using lock = await this.mutex.acquire(
|
||||
`calendar:subscription:${subscriptionId}`
|
||||
);
|
||||
@@ -314,6 +326,12 @@ export class CalendarService {
|
||||
return;
|
||||
}
|
||||
|
||||
const lockedNow = Date.now();
|
||||
const lockedBackoff = await this.getSyncFailureBackoff(subscription.id);
|
||||
if (lockedBackoff && lockedNow < lockedBackoff.nextRetryAt.getTime()) {
|
||||
return;
|
||||
}
|
||||
|
||||
const provider = this.providerFactory.get(
|
||||
account.provider as CalendarProviderName
|
||||
);
|
||||
@@ -321,8 +339,18 @@ export class CalendarService {
|
||||
return;
|
||||
}
|
||||
|
||||
const { accessToken } = await this.ensureAccessToken(account);
|
||||
if (!accessToken) {
|
||||
let accessToken: string | null = null;
|
||||
try {
|
||||
const tokens = await this.ensureAccessToken(account);
|
||||
if (!tokens.accessToken) return;
|
||||
accessToken = tokens.accessToken;
|
||||
} catch (error) {
|
||||
await this.handleSubscriptionSyncFailure({
|
||||
error,
|
||||
subscription,
|
||||
account,
|
||||
provider,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -352,31 +380,42 @@ export class CalendarService {
|
||||
await this.models.calendarSubscription.updateSync(subscription.id, {
|
||||
syncToken: null,
|
||||
});
|
||||
await this.syncWithProvider({
|
||||
provider,
|
||||
subscriptionId: subscription.id,
|
||||
calendarId: subscription.externalCalendarId,
|
||||
accessToken,
|
||||
account,
|
||||
timeMin,
|
||||
timeMax,
|
||||
subscriptionTimezone: subscription.timezone ?? undefined,
|
||||
});
|
||||
synced = true;
|
||||
} else {
|
||||
if (this.isTokenInvalidError(error)) {
|
||||
await this.invalidateAccount(account.id, (error as Error).message);
|
||||
} else {
|
||||
this.logger.warn(
|
||||
`Calendar sync failed for subscription ${subscription.id}`,
|
||||
error as Error
|
||||
);
|
||||
try {
|
||||
await this.syncWithProvider({
|
||||
provider,
|
||||
subscriptionId: subscription.id,
|
||||
calendarId: subscription.externalCalendarId,
|
||||
accessToken,
|
||||
account,
|
||||
timeMin,
|
||||
timeMax,
|
||||
subscriptionTimezone: subscription.timezone ?? undefined,
|
||||
});
|
||||
synced = true;
|
||||
} catch (syncTokenRetryError) {
|
||||
await this.handleSubscriptionSyncFailure({
|
||||
error: syncTokenRetryError,
|
||||
subscription,
|
||||
account,
|
||||
provider,
|
||||
accessToken,
|
||||
});
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
await this.handleSubscriptionSyncFailure({
|
||||
error,
|
||||
subscription,
|
||||
account,
|
||||
provider,
|
||||
accessToken,
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (synced) {
|
||||
await this.clearSyncFailureBackoff(subscription.id);
|
||||
await this.ensureWebhookChannel(subscription, provider, accessToken);
|
||||
}
|
||||
|
||||
@@ -764,19 +803,6 @@ export class CalendarService {
|
||||
return false;
|
||||
}
|
||||
|
||||
private async invalidateAccount(accountId: string, lastError?: string) {
|
||||
await this.models.calendarAccount.updateStatus(
|
||||
accountId,
|
||||
'invalid',
|
||||
lastError ?? null
|
||||
);
|
||||
const subscriptions =
|
||||
await this.models.calendarSubscription.listByAccount(accountId);
|
||||
const subscriptionIds = subscriptions.map(s => s.id);
|
||||
await this.models.calendarEvent.deleteBySubscriptionIds(subscriptionIds);
|
||||
await this.models.calendarSubscription.clearSyncTokensByAccount(accountId);
|
||||
}
|
||||
|
||||
private requireProvider(name: CalendarProviderName) {
|
||||
const provider = this.providerFactory.get(name);
|
||||
if (!provider) {
|
||||
@@ -841,4 +867,180 @@ export class CalendarService {
|
||||
channelExpiration: result.expiration ?? null,
|
||||
});
|
||||
}
|
||||
|
||||
private async handleSubscriptionSyncFailure(params: {
|
||||
error: unknown;
|
||||
subscription: {
|
||||
id: string;
|
||||
externalCalendarId: string;
|
||||
customChannelId: string | null;
|
||||
customResourceId: string | null;
|
||||
};
|
||||
account: CalendarAccount;
|
||||
provider: CalendarProvider;
|
||||
accessToken?: string;
|
||||
}) {
|
||||
if (this.isSubscriptionMissingError(params.error)) {
|
||||
await this.disableSubscription({
|
||||
subscriptionId: params.subscription.id,
|
||||
provider: params.provider,
|
||||
accessToken: params.accessToken,
|
||||
customChannelId: params.subscription.customChannelId,
|
||||
customResourceId: params.subscription.customResourceId,
|
||||
});
|
||||
this.logger.warn(
|
||||
`Calendar subscription ${params.subscription.id} was disabled because provider returned 404 for calendar ${params.subscription.externalCalendarId}`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.isTokenInvalidError(params.error)) {
|
||||
await this.clearSyncFailureBackoff(params.subscription.id);
|
||||
await this.models.calendarAccount.invalidateAndPurge(
|
||||
params.account.id,
|
||||
this.formatSyncError(params.error)
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const backoff = await this.bumpSyncFailureBackoff(params.subscription.id);
|
||||
const interval = params.account.refreshIntervalMinutes ?? 60;
|
||||
const lastSyncAt = this.calculateLastSyncAtForRetry(
|
||||
backoff.nextRetryAt,
|
||||
interval
|
||||
);
|
||||
await this.models.calendarSubscription.updateLastSyncAt(
|
||||
params.subscription.id,
|
||||
lastSyncAt
|
||||
);
|
||||
this.logger.warn(
|
||||
`Calendar sync failed for subscription ${params.subscription.id}, attempt ${backoff.attempt}, next retry at ${backoff.nextRetryAt.toISOString()}`,
|
||||
this.toError(params.error)
|
||||
);
|
||||
}
|
||||
|
||||
private isSubscriptionMissingError(error: unknown) {
|
||||
if (!(error instanceof CalendarProviderRequestError)) {
|
||||
return false;
|
||||
}
|
||||
const status = error.data?.status ?? error.status;
|
||||
return status === 404;
|
||||
}
|
||||
|
||||
private calculateLastSyncAtForRetry(
|
||||
nextRetryAt: Date,
|
||||
refreshIntervalMinutes: number
|
||||
) {
|
||||
// Cron schedules by `now - lastSyncAt >= refreshInterval`, so back-calculate
|
||||
// a synthetic lastSyncAt to defer the next attempt to `nextRetryAt`.
|
||||
return new Date(nextRetryAt.getTime() - refreshIntervalMinutes * 60 * 1000);
|
||||
}
|
||||
|
||||
private async disableSubscription(params: {
|
||||
subscriptionId: string;
|
||||
provider: CalendarProvider;
|
||||
accessToken?: string;
|
||||
customChannelId: string | null;
|
||||
customResourceId: string | null;
|
||||
}) {
|
||||
if (
|
||||
params.provider.stopChannel &&
|
||||
params.accessToken &&
|
||||
params.customChannelId &&
|
||||
params.customResourceId
|
||||
) {
|
||||
try {
|
||||
await params.provider.stopChannel({
|
||||
accessToken: params.accessToken,
|
||||
channelId: params.customChannelId,
|
||||
resourceId: params.customResourceId,
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.warn(
|
||||
`Failed to stop webhook channel for disabled calendar subscription ${params.subscriptionId}`,
|
||||
this.toError(error)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
await this.models.calendarSubscription.disableAndPurge(
|
||||
params.subscriptionId
|
||||
);
|
||||
await this.clearSyncFailureBackoff(params.subscriptionId);
|
||||
}
|
||||
|
||||
private getSyncFailureBackoffKey(subscriptionId: string) {
|
||||
return `${SYNC_FAILURE_BACKOFF_KEY_PREFIX}${subscriptionId}`;
|
||||
}
|
||||
|
||||
private async getSyncFailureBackoff(subscriptionId: string) {
|
||||
const key = this.getSyncFailureBackoffKey(subscriptionId);
|
||||
const value = await this.redis.get(key);
|
||||
if (!value) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
const parsed = JSON.parse(value) as {
|
||||
attempt?: number;
|
||||
nextRetryAt?: string;
|
||||
};
|
||||
if (!parsed.attempt || !parsed.nextRetryAt) {
|
||||
return null;
|
||||
}
|
||||
const nextRetryAt = new Date(parsed.nextRetryAt);
|
||||
if (Number.isNaN(nextRetryAt.getTime())) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
attempt: parsed.attempt,
|
||||
nextRetryAt,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private async bumpSyncFailureBackoff(subscriptionId: string) {
|
||||
const state = await this.getSyncFailureBackoff(subscriptionId);
|
||||
const attempt = (state?.attempt ?? 0) + 1;
|
||||
const delay = Math.min(
|
||||
SYNC_FAILURE_BACKOFF_BASE_MS * 2 ** (attempt - 1),
|
||||
SYNC_FAILURE_BACKOFF_MAX_MS
|
||||
);
|
||||
const nextRetryAt = new Date(Date.now() + delay);
|
||||
const key = this.getSyncFailureBackoffKey(subscriptionId);
|
||||
await this.redis.set(
|
||||
key,
|
||||
JSON.stringify({
|
||||
attempt,
|
||||
nextRetryAt: nextRetryAt.toISOString(),
|
||||
}),
|
||||
'EX',
|
||||
SYNC_FAILURE_BACKOFF_TTL_SECONDS
|
||||
);
|
||||
return {
|
||||
attempt,
|
||||
nextRetryAt,
|
||||
};
|
||||
}
|
||||
|
||||
private async clearSyncFailureBackoff(subscriptionId: string) {
|
||||
const key = this.getSyncFailureBackoffKey(subscriptionId);
|
||||
await this.redis.del(key);
|
||||
}
|
||||
|
||||
private formatSyncError(error: unknown) {
|
||||
if (error instanceof Error && error.message) {
|
||||
return error.message;
|
||||
}
|
||||
return String(error);
|
||||
}
|
||||
|
||||
private toError(error: unknown) {
|
||||
if (error instanceof Error) {
|
||||
return error;
|
||||
}
|
||||
return new Error(this.formatSyncError(error));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user