andras/reconcile
andras/reconcile
1
0
Fork 0
Code Issues Pull requests 6 Projects Releases Packages Wiki Activity Actions

Hoist retry logic

Browse source
This commit is contained in:
Andras Schmelczer 2025-03-22 16:15:33 +00:00
commit c7e53bff26
No known key found for this signature in database
GPG key ID: FC8F2C3D3D1A718C
4 changed files with 215 additions and 189 deletions
Download patch file Download diff file Expand all files Collapse all files

1
frontend/obsidian-plugin/src/views/settings-tab.ts
View file

@ -211,6 +211,7 @@ export class SyncSettingsTab extends PluginSettingTab {
new Notice( new Notice(
"The changes have been applied successfully!" "The changes have been applied successfully!"
); );
await this.statusDescription.updateConnectionState();
} else { } else {
new Notice("No changes to apply"); new Notice("No changes to apply");
} }

70
frontend/sync-client/src/services/connection-status.ts
View file

@ -39,65 +39,53 @@ export class ConnectionStatus {
return input.url; return input.url;
} }
public getFetchImplementation(
fetch: typeof globalThis.fetch,
{ doRetries = true }: { doRetries: boolean } = { doRetries: true }
): typeof globalThis.fetch {
return doRetries ? this.retriedFetchFactory(this.logger, fetch) : fetch;
}
public reset(): void { public reset(): void {
this.rejectUntil(new Error("Sync was reset")); this.rejectUntil(new Error("Sync was reset"));
[this.until, this.resolveUntil, this.rejectUntil] = createPromise(); [this.until, this.resolveUntil, this.rejectUntil] = createPromise();
} }
private retriedFetchFactory( public getFetchImplementation(
logger: Logger, logger: Logger,
fetch: typeof globalThis.fetch = globalThis.fetch fetch: typeof globalThis.fetch = globalThis.fetch
): typeof globalThis.fetch { ): typeof globalThis.fetch {
return async (input: RequestInfo | URL): Promise<Response> => { return async (input: RequestInfo | URL): Promise<Response> => {
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition while (!this.canFetch) {
while (true) { await this.until;
while (!this.canFetch) { }
await this.until;
}
try { try {
// https://github.com/jonbern/fetch-retry/blob/8684ef4e688375f623bd76f13add76dbc1d67cfb/index.js#L67C1-L70C21 // https://github.com/jonbern/fetch-retry/blob/8684ef4e688375f623bd76f13add76dbc1d67cfb/index.js#L67C1-L70C21
const _input = const _input =
typeof Request !== "undefined" && typeof Request !== "undefined" && input instanceof Request
input instanceof Request ? input.clone()
? input.clone() : input;
: input;
const fetchPromise = fetch(_input); const fetchPromise = fetch(_input);
// We only want to catch rejections from `this.until` // We only want to catch rejections from `this.until`
let result: symbol | Response | undefined = undefined; let result: symbol | Response | undefined = undefined;
do { do {
result = await Promise.race([this.until, fetchPromise]); result = await Promise.race([this.until, fetchPromise]);
} while (result === ConnectionStatus.UNTIL_RESOLUTION); } while (result === ConnectionStatus.UNTIL_RESOLUTION);
const fetchResult: Response = result as Response; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion const fetchResult: Response = result as Response; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
if (!fetchResult.ok) { if (!fetchResult.ok) {
this.logger.warn( this.logger.warn(
`Retrying fetch for ${ConnectionStatus.getUrlFromInput( `Fetch for ${ConnectionStatus.getUrlFromInput(
input
)}, got status ${fetchResult.status}`
);
}
return fetchResult;
} catch (error) {
logger.warn(
`Retrying fetch for ${ConnectionStatus.getUrlFromInput(
input input
)}, got error: ${error}` )}, got status ${fetchResult.status}`
); );
} }
await Promise.race([this.until, sleep(1000)]); return fetchResult;
} catch (error) {
logger.warn(
`Fetch for ${ConnectionStatus.getUrlFromInput(
input
)}, got error: ${error}`
);
throw error;
} }
}; };
} }

335
frontend/sync-client/src/services/sync-service.ts
View file

@ -9,6 +9,7 @@ import type {
import type { Logger } from "../tracing/logger"; import type { Logger } from "../tracing/logger";
import type { Settings } from "../persistence/settings"; import type { Settings } from "../persistence/settings";
import type { ConnectionStatus } from "./connection-status"; import type { ConnectionStatus } from "./connection-status";
import { sleep } from "../utils/sleep";
export interface CheckConnectionResult { export interface CheckConnectionResult {
isSuccessful: boolean; isSuccessful: boolean;
@ -16,8 +17,8 @@ export interface CheckConnectionResult {
} }
export class SyncService { export class SyncService {
private client!: Client<paths>; private client: Client<paths>;
private clientWithoutRetries!: Client<paths>; private pingClient: Client<paths>;
private _fetchImplementation: typeof globalThis.fetch = globalThis.fetch; private _fetchImplementation: typeof globalThis.fetch = globalThis.fetch;
public constructor( public constructor(
@ -25,20 +26,26 @@ export class SyncService {
private readonly settings: Settings, private readonly settings: Settings,
private readonly logger: Logger private readonly logger: Logger
) { ) {
this.createClient(this.settings.getSettings().remoteUri); [this.client, this.pingClient] = this.createClient(
this.settings.getSettings().remoteUri
);
settings.addOnSettingsChangeListener((newSettings, oldSettings) => { settings.addOnSettingsChangeListener((newSettings, oldSettings) => {
if (newSettings.remoteUri === oldSettings.remoteUri) { if (newSettings.remoteUri === oldSettings.remoteUri) {
return; return;
} }
this.createClient(newSettings.remoteUri); [this.client, this.pingClient] = this.createClient(
newSettings.remoteUri
);
}); });
} }
public set fetchImplementation(fetch: typeof globalThis.fetch) { public set fetchImplementation(fetch: typeof globalThis.fetch) {
this._fetchImplementation = fetch; this._fetchImplementation = fetch;
this.createClient(this.settings.getSettings().remoteUri); [this.client, this.pingClient] = this.createClient(
this.settings.getSettings().remoteUri
);
} }
private static formatError( private static formatError(
@ -62,42 +69,44 @@ export class SyncService {
relativePath: RelativePath; relativePath: RelativePath;
contentBytes: Uint8Array; contentBytes: Uint8Array;
}): Promise<components["schemas"]["DocumentVersionWithoutContent"]> { }): Promise<components["schemas"]["DocumentVersionWithoutContent"]> {
const formData = new FormData(); return this.withRetries(async () => {
if (documentId !== undefined) { const formData = new FormData();
formData.append("document_id", documentId); if (documentId !== undefined) {
} formData.append("document_id", documentId);
formData.append("relative_path", relativePath);
formData.append("content", new Blob([contentBytes]));
const response = await this.client.POST(
"/vaults/{vault_id}/documents",
{
params: {
path: {
vault_id: this.settings.getSettings().vaultName
},
header: {
authorization: `Bearer ${this.settings.getSettings().token}`
}
},
// eslint-disable-next-line
body: formData as any // FormData is not supported by openapi-fetch
} }
); formData.append("relative_path", relativePath);
formData.append("content", new Blob([contentBytes]));
if (!response.data) { const response = await this.client.POST(
throw new Error( "/vaults/{vault_id}/documents",
`Failed to create document: ${SyncService.formatError(response.error)}` {
params: {
path: {
vault_id: this.settings.getSettings().vaultName
},
header: {
authorization: `Bearer ${this.settings.getSettings().token}`
}
},
// eslint-disable-next-line
body: formData as any // FormData is not supported by openapi-fetch
}
); );
}
this.logger.debug( if (!response.data) {
`Created document ${JSON.stringify(response.data)} with id ${ throw new Error(
response.data.documentId `Failed to create document: ${SyncService.formatError(response.error)}`
}` );
); }
return response.data; this.logger.debug(
`Created document ${JSON.stringify(response.data)} with id ${
response.data.documentId
}`
);
return response.data;
});
} }
public async put({ public async put({
@ -111,44 +120,46 @@ export class SyncService {
relativePath: RelativePath; relativePath: RelativePath;
contentBytes: Uint8Array; contentBytes: Uint8Array;
}): Promise<components["schemas"]["DocumentUpdateResponse"]> { }): Promise<components["schemas"]["DocumentUpdateResponse"]> {
this.logger.debug( return this.withRetries(async () => {
`Updating document ${documentId} with parent version ${parentVersionId} and relative path ${relativePath}` this.logger.debug(
); `Updating document ${documentId} with parent version ${parentVersionId} and relative path ${relativePath}`
const formData = new FormData();
formData.append("parent_version_id", parentVersionId.toString());
formData.append("relative_path", relativePath);
formData.append("content", new Blob([contentBytes]));
const response = await this.client.PUT(
"/vaults/{vault_id}/documents/{document_id}",
{
params: {
path: {
vault_id: this.settings.getSettings().vaultName,
document_id: documentId
},
header: {
authorization: `Bearer ${this.settings.getSettings().token}`
}
},
// eslint-disable-next-line
body: formData as any // FormData is not supported by openapi-fetch
}
);
if (!response.data) {
throw new Error(
`Failed to update document: ${SyncService.formatError(response.error)}`
); );
} const formData = new FormData();
formData.append("parent_version_id", parentVersionId.toString());
formData.append("relative_path", relativePath);
formData.append("content", new Blob([contentBytes]));
this.logger.debug( const response = await this.client.PUT(
`Updated document ${JSON.stringify(response.data)} with id ${ "/vaults/{vault_id}/documents/{document_id}",
response.data.documentId {
}` params: {
); path: {
vault_id: this.settings.getSettings().vaultName,
document_id: documentId
},
header: {
authorization: `Bearer ${this.settings.getSettings().token}`
}
},
// eslint-disable-next-line
body: formData as any // FormData is not supported by openapi-fetch
}
);
return response.data; if (!response.data) {
throw new Error(
`Failed to update document: ${SyncService.formatError(response.error)}`
);
}
this.logger.debug(
`Updated document ${JSON.stringify(response.data)} with id ${
response.data.documentId
}`
);
return response.data;
});
} }
public async delete({ public async delete({
@ -158,33 +169,35 @@ export class SyncService {
documentId: DocumentId; documentId: DocumentId;
relativePath: RelativePath; relativePath: RelativePath;
}): Promise<components["schemas"]["DocumentVersionWithoutContent"]> { }): Promise<components["schemas"]["DocumentVersionWithoutContent"]> {
const response = await this.client.DELETE( return this.withRetries(async () => {
"/vaults/{vault_id}/documents/{document_id}", const response = await this.client.DELETE(
{ "/vaults/{vault_id}/documents/{document_id}",
params: { {
path: { params: {
vault_id: this.settings.getSettings().vaultName, path: {
document_id: documentId vault_id: this.settings.getSettings().vaultName,
document_id: documentId
},
header: {
authorization: `Bearer ${this.settings.getSettings().token}`
}
}, },
header: { body: {
authorization: `Bearer ${this.settings.getSettings().token}` relativePath
} }
},
body: {
relativePath
} }
);
if (response.error) {
throw new Error(`Failed to delete document`);
} }
);
if (response.error) { this.logger.debug(
throw new Error(`Failed to delete document`); `Deleted document ${relativePath} with id ${documentId}`
} );
this.logger.debug( return response.data;
`Deleted document ${relativePath} with id ${documentId}` });
);
return response.data;
} }
public async get({ public async get({
@ -192,63 +205,70 @@ export class SyncService {
}: { }: {
documentId: DocumentId; documentId: DocumentId;
}): Promise<components["schemas"]["DocumentVersion"]> { }): Promise<components["schemas"]["DocumentVersion"]> {
const response = await this.client.GET( return this.withRetries(async () => {
"/vaults/{vault_id}/documents/{document_id}", const response = await this.client.GET(
{ "/vaults/{vault_id}/documents/{document_id}",
params: { {
path: { params: {
vault_id: this.settings.getSettings().vaultName, path: {
document_id: documentId vault_id: this.settings.getSettings().vaultName,
}, document_id: documentId
header: { },
authorization: `Bearer ${this.settings.getSettings().token}` header: {
authorization: `Bearer ${this.settings.getSettings().token}`
}
} }
} }
}
);
if (!response.data) {
throw new Error(
`Failed to get document: ${SyncService.formatError(response.error)}`
); );
}
this.logger.debug( if (!response.data) {
`Get document ${response.data.relativePath} with id ${response.data.documentId}` throw new Error(
); `Failed to get document: ${SyncService.formatError(response.error)}`
);
}
return response.data; this.logger.debug(
`Get document ${response.data.relativePath} with id ${response.data.documentId}`
);
return response.data;
});
} }
public async getAll( public async getAll(
since?: VaultUpdateId since?: VaultUpdateId
): Promise<components["schemas"]["FetchLatestDocumentsResponse"]> { ): Promise<components["schemas"]["FetchLatestDocumentsResponse"]> {
const response = await this.client.GET("/vaults/{vault_id}/documents", { return this.withRetries(async () => {
params: { const response = await this.client.GET(
path: { "/vaults/{vault_id}/documents",
vault_id: this.settings.getSettings().vaultName {
}, params: {
header: { path: {
authorization: `Bearer ${this.settings.getSettings().token}` vault_id: this.settings.getSettings().vaultName
}, },
query: { header: {
since_update_id: since authorization: `Bearer ${this.settings.getSettings().token}`
},
query: {
since_update_id: since
}
}
} }
}
});
const { error } = response;
if (error) {
throw new Error(
`Failed to get documents: ${SyncService.formatError(response.error)}`
); );
}
this.logger.debug( const { error } = response;
`Got ${response.data.latestDocuments.length} document metadata` if (error) {
); throw new Error(
`Failed to get documents: ${SyncService.formatError(response.error)}`
);
}
return response.data; this.logger.debug(
`Got ${response.data.latestDocuments.length} document metadata`
);
return response.data;
});
} }
public async checkConnection(): Promise<CheckConnectionResult> { public async checkConnection(): Promise<CheckConnectionResult> {
@ -273,8 +293,9 @@ export class SyncService {
} }
} }
// No retries
private async ping(): Promise<components["schemas"]["PingResponse"]> { private async ping(): Promise<components["schemas"]["PingResponse"]> {
const response = await this.clientWithoutRetries.GET("/ping", { const response = await this.pingClient.GET("/ping", {
params: { params: {
header: { header: {
authorization: `Bearer ${this.settings.getSettings().token}` authorization: `Bearer ${this.settings.getSettings().token}`
@ -293,20 +314,34 @@ export class SyncService {
return response.data; return response.data;
} }
private createClient(remoteUri: string): void { /**
this.client = createClient<paths>({ * Create a client and a ping client for the given remote URI.
baseUrl: remoteUri, */
fetch: this.connectionStatus.getFetchImplementation( private createClient(remoteUri: string): [Client<paths>, Client<paths>] {
this._fetchImplementation return [
) createClient<paths>({
}); baseUrl: remoteUri,
fetch: this.connectionStatus.getFetchImplementation(
this.logger,
this._fetchImplementation
)
}),
createClient<paths>({
baseUrl: remoteUri,
fetch: this._fetchImplementation
})
];
}
this.clientWithoutRetries = createClient<paths>({ private async withRetries<T>(fn: () => Promise<T>): Promise<T> {
baseUrl: remoteUri, // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
fetch: this.connectionStatus.getFetchImplementation( while (true) {
this._fetchImplementation, try {
{ doRetries: false } return await fn();
) } catch (e) {
}); this.logger.error(`Failed network call (${e}), retrying`);
await sleep(1000);
}
}
} }
} }

6
frontend/sync-client/src/sync-operations/unrestricted-syncer.ts
View file

@ -177,7 +177,9 @@ export class UnrestrictedSyncer {
} }
if ( if (
document.metadata.parentVersionId >= response.vaultUpdateId // `Syncer` creates fake local document metadata for all remote docs with invalid hashes. The parent IDs will likely match
// the latest versions so we still need to update the local versions to turn the fakes into real metadata.
document.metadata.parentVersionId > response.vaultUpdateId
) { ) {
this.logger.debug( this.logger.debug(
`Document ${document.relativePath} is already more up to date than the fetched version` `Document ${document.relativePath} is already more up to date than the fetched version`
@ -281,7 +283,7 @@ export class UnrestrictedSyncer {
remoteVersion.vaultUpdateId remoteVersion.vaultUpdateId
) { ) {
this.logger.debug( this.logger.debug(
`Document ${remoteVersion.relativePath} is already more up to date than the fetched version` `Document ${remoteVersion.relativePath} is already at least as up to date as the fetched version`
); );
return; return;
} }