fix(docsync): stop a push from stealing another file's pairing

Papra answers a push of bytes it already holds with the document that has
them. Two trip files with the same content therefore pushed to one document,
and the unique index on (link_id, remote_id) aborted that run and every run
after it with a constraint error. The pairing now stays with the file that
owns it and the second file is recorded as a conflict.

Also in this pass:
- Paperless no longer adopts a same-bytes document that lacks the scope's tag
  when a consume outlives the poll budget; that document belongs to someone
  else's filing and delete-through would have trashed it.
- The pull path writes the file row, the retirement of the copy it replaces
  and the pairing in one transaction, and announces them only once committed.
- PATCH links stops accepting remoteRootId/remoteRootPath, which it read and
  silently dropped.
- The MCP sync tool clears shelved items before running, as the REST route
  does.
This commit is contained in:
Maurice
2026-09-19 00:44:43 +02:00
parent 7237a1881d
commit e6a5ca1521
11 changed files with 267 additions and 39 deletions
@@ -100,14 +100,20 @@ export class DocSyncWebhookController implements OnModuleDestroy {
* up again when the timer fires, so a binding switched off or deleted in the
* meantime does not get one last run out of a stale row.
*/
private schedule(linkId: number, reload: () => ReturnType<DocSyncConfigService['getLink']>): void {
private schedule(linkId: number, reload: () => ReturnType<DocSyncConfigService['getLink']>, isRetry = false): void {
if (this.pending.has(linkId)) return;
const timer = setTimeout(() => {
this.pending.delete(linkId);
const fresh = reload();
if (!fresh || fresh.sync_enabled !== 1) return;
if (!this.syncIsOn()) return;
void this.sync.syncLink(fresh);
void this.sync.syncLink(fresh).then((res) => {
// A run that was already in flight answers `busy`, and the changes this
// nudge was about may have landed after that run read the folder. Ask
// again once rather than waiting out a whole poll interval — once, and
// only for `busy`, so this cannot become a loop.
if (res?.state === 'busy' && !isRetry) this.schedule(linkId, reload, true);
});
}, WEBHOOK_NUDGE_DEBOUNCE_SECONDS * 1000);
// A pending nudge must not hold the process open at shutdown.
if (typeof timer.unref === 'function') timer.unref();
+8 -3
View File
@@ -329,9 +329,14 @@ export function planReconcile(input: ReconcileInput): ReconcilePlan {
// `both` as "TREK always wins" silently renamed the provider's copy back
// every time somebody tidied up a folder, which is the opposite of a
// two-way sync.
if (l && r.name !== l.name) {
const providerRenamed = it.remoteName !== null && r.name !== it.remoteName;
const localRenamed = it.remoteName !== null && l.name !== it.remoteName;
if (l && sanitizeIncomingName(r.name) !== l.name) {
// Compared through the same sanitiser the local name went through, or a
// provider name TREK had to clean up (a slash, a control character, a
// leading dot) reads as "TREK renamed this" on the very next run — and
// the provider's copy gets renamed to the cleaned version, unasked.
const agreed = it.remoteName === null ? null : sanitizeIncomingName(it.remoteName);
const providerRenamed = agreed !== null && sanitizeIncomingName(r.name) !== agreed;
const localRenamed = agreed !== null && l.name !== agreed;
if (providerRenamed && !localRenamed && direction !== 'push') {
add({ kind: 'rename_local', itemId: it.id, fileId: l.fileId, name: r.name });
+5
View File
@@ -97,6 +97,11 @@ export class DocSyncMcp {
}
const results = [];
for (const link of links) {
// Asking for a run by hand means "try again", including the documents
// that were shelved after too many failures. The REST route does the
// same thing before its run; a tool that skipped it would answer "in
// sync" while leaving them shelved.
this.sync.retryShelvedItems(link.id);
results.push({ linkId: link.id, provider: link.provider_id, ...(await this.sync.syncLink(link, { full: full === true })) });
}
return ok({ runs: results });
+80 -31
View File
@@ -7,6 +7,8 @@ import { Transform } from 'node:stream';
import { Readable } from 'node:stream';
import type { DocsyncErrorCode } from '@trek/shared';
import { DatabaseService } from '../database/database.service';
import { AddonsService } from '../addons/addons.service';
import type { AddonId } from '../../addons';
import { StorageService } from '../storage/storage.service';
import { FilesService } from '../files/files.service';
import { AllowedFileTypesService } from '../files/allowed-file-types.service';
@@ -70,6 +72,7 @@ export class DocSyncService {
private readonly files: FilesService,
private readonly allowedTypes: AllowedFileTypesService,
private readonly realtime: RealtimeService,
private readonly addons: AddonsService,
) {}
// ── Entry points ───────────────────────────────────────────────────────────
@@ -143,6 +146,13 @@ export class DocSyncService {
this.recordLinkFailure(link, 'provider_error');
return { state: 'failed', pulled: 0, pushed: 0, conflicts: 0, missing: 0, errorCode: 'provider_error' };
}
// Switching a provider off in the admin panel stopped new connections and
// left every existing binding running, which is not what "off" means to the
// person who switched it. Not recorded as a failure: nothing is wrong with
// the binding, an admin has closed the door.
if (!this.addons.isAddonEnabled(link.provider_id as AddonId)) {
return { state: 'disabled', pulled: 0, pushed: 0, conflicts: 0, missing: 0 };
}
const ref = this.config.toRef(conn);
const scope = this.config.toScopeRef(link);
@@ -371,39 +381,42 @@ export class DocSyncService {
.prepare('SELECT file_id FROM document_sync_items WHERE id = ?')
.get(itemId) as { file_id: number | null } | undefined)?.file_id ?? null;
const created = this.files.createFile(
ctx.link.trip_id,
{ filename: storageKey, originalname: name, size: bytes, mimetype: remote.mimeType || 'application/octet-stream' },
// Attributed to the person whose connection brought it in, which is the
// only honest answer: nobody in TREK uploaded it.
this.config.getConnection(ctx.link.connection_id)?.owner_user_id ?? 0,
{},
);
/**
* Retire the copy this one replaces.
* The file row, the retirement of the copy it replaces and the pairing are
* one fact, so they are one transaction. Written separately, a crash
* between them leaves a trip file with no sync item — which the next run
* reads as a document TREK gained and pushes straight back up, turning one
* upstream edit into two documents on both sides.
*
* Into the trash rather than out of existence: the bytes it holds are a
* version somebody may still want, and TREK's own delete works the same way.
* Leaving it in place was worse than untidy — the row's pairing moves to the
* new file, so the old one is left with no sync item at all, and the very
* next run reads it as a document TREK gained and pushes the superseded
* version back up. One upstream edit then ends as two documents on both
* sides, and it grows with every further edit.
* The superseded copy goes to the trash rather than out of existence: the
* bytes it holds are a version somebody may still want, and TREK's own
* delete works the same way.
*/
if (supersededId !== null && Number(supersededId) !== Number(created.id)) {
this.files.softDeleteFile(supersededId);
this.realtime.broadcast(ctx.link.trip_id, 'file:deleted', { fileId: supersededId });
}
this.upsertItem(ctx.link, {
itemId,
remote,
state: 'synced',
fileId: Number(created.id),
contentSha256: sha256,
errorCode: null,
const { created, retired } = this.db.transaction(() => {
const file = this.files.createFile(
ctx.link.trip_id,
{ filename: storageKey, originalname: name, size: bytes, mimetype: remote.mimeType || 'application/octet-stream' },
// Attributed to the person whose connection brought it in, which is the
// only honest answer: nobody in TREK uploaded it.
this.config.getConnection(ctx.link.connection_id)?.owner_user_id ?? 0,
{},
);
const gone = supersededId !== null && Number(supersededId) !== Number(file.id);
if (gone) this.files.softDeleteFile(supersededId as number);
this.upsertItem(ctx.link, {
itemId,
remote,
state: 'synced',
fileId: Number(file.id),
contentSha256: sha256,
errorCode: null,
});
return { created: file, retired: gone ? (supersededId as number) : null };
});
// Announced only once it is committed — a rolled-back transaction that had
// already told every client the file exists cannot be taken back.
if (retired !== null) this.realtime.broadcast(ctx.link.trip_id, 'file:deleted', { fileId: retired });
this.realtime.broadcast(ctx.link.trip_id, 'file:created', { file: created });
return 'pulled';
}
@@ -469,6 +482,24 @@ export class DocSyncService {
return res.error.code;
}
// A provider may answer a push with a document it already held rather than
// a new one: Papra deduplicates identical bytes, and `deduplicated` says so.
// Two trip files with the same content therefore push to ONE document, and
// the unique index on (link_id, remote_id) would abort this run — and every
// run after it — with a constraint error. The upload happened, so nothing is
// lost; what must not happen is the pairing moving off the file that owns it.
const heldBy = this.pairingOwner(ctx.link.id, res.data.remoteId, itemId);
if (heldBy !== null) {
this.upsertItem(ctx.link, {
itemId,
fileId: local.fileId,
state: 'error',
errorCode: 'conflict',
trekDocUid: uid,
});
return 'conflict';
}
this.upsertItem(ctx.link, {
itemId,
fileId: local.fileId,
@@ -477,6 +508,12 @@ export class DocSyncService {
remoteVersion: res.data.remoteVersion,
contentSha256: sha256,
pushedSha256: sha256,
// The name both sides now agree on. Leaving it null meant the first
// `touch` filled the arbiter with whatever the provider had made of the
// name — Paperless stores a title and keeps its own filename — and the
// run after that read the difference as TREK having renamed the document
// and renamed the provider's copy to match, unasked.
remoteNameOverride: local.name,
trekDocUid: uid,
errorCode: null,
});
@@ -567,6 +604,15 @@ export class DocSyncService {
}));
}
/** The item already paired with this remote document, if it is a different one. */
private pairingOwner(linkId: number, remoteId: string, itemId: number | null): number | null {
const row = this.db.connection
.prepare('SELECT id FROM document_sync_items WHERE link_id = ? AND remote_id = ?')
.get(linkId, remoteId) as { id?: number } | undefined;
if (row?.id === undefined) return null;
return itemId !== null && Number(row.id) === Number(itemId) ? null : Number(row.id);
}
private existingUid(linkId: number, itemId: number | null): string | null {
if (itemId === null) return null;
const row = this.db.connection
@@ -588,6 +634,8 @@ export class DocSyncService {
itemId: number | null;
remote?: RemoteDocument;
remoteIdOverride?: string;
/** The agreed name, where the caller knows it better than the listing does. */
remoteNameOverride?: string;
remoteVersion?: string;
fileId?: number;
state: string;
@@ -599,6 +647,7 @@ export class DocSyncService {
): void {
const remoteId = patch.remoteIdOverride ?? patch.remote?.remoteId ?? null;
const remoteVersion = patch.remoteVersion ?? patch.remote?.remoteVersion ?? null;
const remoteName = patch.remoteNameOverride ?? patch.remote?.name ?? null;
const failed = patch.state === 'error';
if (patch.itemId !== null) {
@@ -644,7 +693,7 @@ export class DocSyncService {
.run(
patch.state, patch.errorCode ?? null, patch.fileId ?? null,
remoteId, remoteVersion,
patch.remote?.name ?? null, patch.remote?.size ?? null, patch.remote?.remoteModifiedAt ?? null,
remoteName, patch.remote?.size ?? null, patch.remote?.remoteModifiedAt ?? null,
patch.contentSha256 ?? null, patch.pushedSha256 ?? null,
attempts,
failed ? 1 : 0, attempts, ITEM_MAX_ATTEMPTS,
@@ -677,7 +726,7 @@ export class DocSyncService {
)
.run(
link.id, link.trip_id, patch.fileId ?? null, patch.trekDocUid ?? newTrekDocUid(),
remoteId, patch.remote?.name ?? null, remoteVersion, patch.remote?.size ?? null,
remoteId, remoteName, remoteVersion, patch.remote?.size ?? null,
patch.remote?.remoteModifiedAt ?? null, patch.contentSha256 ?? null, patch.pushedSha256 ?? null,
patch.state, patch.errorCode ?? null, failed ? 1 : 0,
failed ? 1 : 0, backoffSeconds(ITEM_BACKOFF_SECONDS, 1), // a new row is always at its first failure
@@ -542,7 +542,12 @@ export class PaperlessDocumentProvider implements DocumentProvider {
// because Paperless accepts duplicate bytes. So the checksum decides.
if (err instanceof PaperlessError && err.code === 'timeout') {
const existing = await this.client.findByChecksum(creds, req.sha256);
const mine = existing.find((doc) => doc.tagIds.includes(tagId)) ?? existing[0];
// Only a document carrying this scope's tag can be the one this push
// made: the tag goes on at upload. Anything else with the same bytes
// belongs to someone else's filing, and adopting it would put a
// stranger's document in the trip — and delete it on the next
// delete-through.
const mine = existing.find((doc) => doc.tagIds.includes(tagId));
if (mine !== undefined) return mine.id;
}
throw err;
@@ -371,3 +371,40 @@ describe('the admin switches', () => {
expect(sync.syncLink).toHaveBeenCalledTimes(1);
});
});
/**
* A nudge that lands on a run already in flight.
*
* `syncLink` answers `busy` and returns, so the nudge was thrown away — and the
* changes it was about may well have landed after the running pass read the
* folder, which means waiting out a whole poll interval for them. Asked again
* once, and only once, so two clients cannot keep each other going.
*/
describe('a nudge that arrives mid-run', () => {
it('asks again once when the run was busy', async () => {
sync.syncLink.mockResolvedValueOnce({ state: 'busy', pulled: 0, pushed: 0, conflicts: 0, missing: 0 });
controller.nudge('tok-live', makeReq({ 'x-trek-docsync-secret': SECRET }));
settle();
await Promise.resolve();
settle();
expect(sync.syncLink).toHaveBeenCalledTimes(2);
});
it('gives up after that one retry rather than chasing itself', async () => {
sync.syncLink.mockResolvedValue({ state: 'busy', pulled: 0, pushed: 0, conflicts: 0, missing: 0 });
controller.nudge('tok-live', makeReq({ 'x-trek-docsync-secret': SECRET }));
for (let i = 0; i < 5; i += 1) { settle(); await Promise.resolve(); }
expect(sync.syncLink).toHaveBeenCalledTimes(2);
});
it('does not retry a run that worked', async () => {
// Set explicitly: clearAllMocks drops the calls but keeps the
// implementation the previous case installed.
sync.syncLink.mockResolvedValue({ state: 'ok', pulled: 0, pushed: 0, conflicts: 0, missing: 0 });
controller.nudge('tok-live', makeReq({ 'x-trek-docsync-secret': SECRET }));
settle();
await Promise.resolve();
settle();
expect(sync.syncLink).toHaveBeenCalledTimes(1);
});
});
@@ -570,3 +570,31 @@ describe('planReconcile > a pairing whose download never landed', () => {
expect(p.actions.map(a => a.kind)).toEqual(['touch']);
});
});
/**
* Names are compared through the same sanitiser the local one went through.
*
* A provider name TREK had to clean up — a slash, a control character, a
* leading dot — otherwise reads as "TREK renamed this document" on the very
* next run, and the provider's copy is renamed to the cleaned version without
* anybody asking for it.
*/
describe('planReconcile > a provider name TREK had to clean up', () => {
it('does not read the cleaning as a local rename', () => {
const p = plan({
items: [item({ remoteName: 'a/b.pdf', contentSha256: 'aaa' })],
remote: [remote({ remoteId: 'r1', name: 'a/b.pdf', remoteVersion: 'v1' })],
local: [local({ fileId: 1, name: 'b.pdf' })], // what sanitizeIncomingName made of it
});
expect(p.actions.map(a => a.kind)).toEqual(['touch']);
});
it('still follows a real rename at the provider', () => {
const p = plan({
items: [item({ remoteName: 'b.pdf', contentSha256: 'aaa' })],
remote: [remote({ remoteId: 'r1', name: 'invoice.pdf', remoteVersion: 'v1' })],
local: [local({ fileId: 1, name: 'b.pdf' })],
});
expect(p.actions.map(a => a.kind)).toEqual(['rename_local']);
});
});
@@ -49,6 +49,7 @@ function makeMcp(over: Partial<Setup> = {}) {
status: vi.fn(() => setup.status),
issues: vi.fn(() => setup.issues),
syncLink: vi.fn(async () => RUN),
retryShelvedItems: vi.fn(),
};
const addons = { isAddonEnabled: vi.fn(() => setup.addonOn) };
const mcp = new DocSyncMcp(
@@ -220,6 +221,16 @@ describe('sync_trip_documents', () => {
]);
});
it('un-shelves the documents that gave up before running, as the REST route does', async () => {
// "Run it now" means "try again", including the items parked after too many
// failures. A tool that skipped this reported "in sync" while leaving them
// parked, and the only way out was the button in the web UI.
const links = [link(1, 'paperless'), link(2, 'nextcloud')];
const { mcp, sync } = makeMcp({ links });
await mcp.syncNow({ tripId: 3 }, ctx);
expect(sync.retryShelvedItems.mock.calls).toEqual([[1], [2]]);
});
it('asks for the cheap incremental run unless the caller says otherwise', async () => {
const { mcp, sync } = makeMcp({ links: [link(1)] });
await mcp.syncNow({ tripId: 3 }, ctx);
@@ -66,6 +66,7 @@ import type {
PushResult,
RemoteDocument,
} from '../../../../src/nest/doc-sync/document-provider';
import type { AddonsService } from '../../../../src/nest/addons/addons.service';
import type { FilesService } from '../../../../src/nest/files/files.service';
import type { StorageService } from '../../../../src/nest/storage/storage.service';
import type { RealtimeService } from '../../../../src/nest/realtime/realtime.service';
@@ -272,6 +273,12 @@ const fileRows = () =>
const sqlTime = (modifier: string): string =>
(testDb.prepare("SELECT datetime('now', ?) AS t").get(modifier) as { t: string }).t;
/**
* The provider's own addon switch. Every binding asks it before it runs, so a
* double that always says yes would hide the one case worth a test.
*/
const addons = { isAddonEnabled: vi.fn(() => true) };
// ── Suite ────────────────────────────────────────────────────────────────────
describe('DocSyncService', () => {
@@ -293,7 +300,16 @@ describe('DocSyncService', () => {
const dbs = new DatabaseService(testDb);
const registry = new DocumentProviderRegistry([provider as unknown as DocumentProvider]);
config = new DocSyncConfigService(dbs, registry);
service = new DocSyncService(dbs, config, registry, storage, files, new AllowedFileTypesService(dbs), realtime);
service = new DocSyncService(
dbs,
config,
registry,
storage,
files,
new AllowedFileTypesService(dbs),
realtime,
addons as unknown as AddonsService,
);
});
afterAll(() => {
@@ -548,6 +564,49 @@ describe('DocSyncService', () => {
expect(itemRows()[0].state).toBe('synced');
});
it('refuses to move a pairing onto a document another file already owns', async () => {
// Papra answers a push of bytes it already holds with the document that
// has them. Two trip files with the same content therefore push to ONE
// document — and the unique index on (link_id, remote_id) turned that
// into a constraint error that aborted this run and every run after it.
const link = makeLink();
makeFile({ name: 'boarding.pdf' });
makeFile({ name: 'boarding-copy.pdf', storageKey: 'key-copy' });
provider.list.mockResolvedValue(ok(listing([])));
const run = await service.syncLink(link);
// Counted as a conflict, which is what it is — a document two files want
// — and reported the same way as any other conflict rather than as a
// failed run.
expect(run.pushed).toBe(1);
expect(run.conflicts).toBe(1);
const rows = itemRows();
expect(rows).toHaveLength(2);
// The first file keeps the document. The second is on record as the one
// that could not have it, with its bytes still in TREK.
expect(rows.filter((r) => r.remote_id === 'r-pushed')).toHaveLength(1);
const blocked = rows.find((r) => r.state === 'error');
expect(blocked?.error_code).toBe('conflict');
expect(blocked?.remote_id).toBeNull();
});
it('stands down while the provider is switched off in the admin panel', async () => {
// Switching a provider off stopped new connections and left every
// existing binding running, which is not what "off" means to the person
// who switched it. Not a failure either: nothing is wrong with the
// binding, an admin has closed the door.
const link = makeLink();
addons.isAddonEnabled.mockReturnValueOnce(false);
const res = await service.syncLink(link);
expect(res.state).toBe('disabled');
expect(provider.list).not.toHaveBeenCalled();
expect(linkRow(link.id).failure_count).toBe(0);
});
it('counts a failure and puts the binding off until later', async () => {
const link = makeLink();
provider.list.mockResolvedValueOnce(fail('unreachable'));
@@ -743,6 +743,20 @@ describe('PaperlessProvider — pushing bytes', () => {
expect(lastRequest('GET /api/documents/').url.searchParams.get('checksum__iexact')).toBe(HASH_A);
});
it('PAPERLESS-057a: a document with the same bytes but another tag is not adopted', async () => {
// The tag goes on at upload, so only a tagged document can be the one this
// push made. Taking the untagged twin instead put a stranger's document in
// the trip — and delete-through would have trashed it on a later run.
vi.spyOn(client, 'awaitConsume').mockRejectedValue(
new PaperlessError('timeout', 'Paperless is still processing the uploaded file'),
);
on('GET /api/custom_fields/', reply(page([{ id: 1, name: TRIP_UID_FIELD_NAME, data_type: 'string' }])));
on('POST /api/documents/post_document/', reply('"task-uuid-1"'));
on('GET /api/documents/', reply(page([docRow({ id: 21, tags: [99] })])));
expect(expectFail(await provider.push(CONN, SCOPE, pushRequest())).code).toBe('timeout');
expect(requests('GET /api/documents/21/')).toHaveLength(0);
});
it('PAPERLESS-057: when the checksum finds nothing either, the timeout is reported', async () => {
vi.spyOn(client, 'awaitConsume').mockRejectedValue(
new PaperlessError('timeout', 'Paperless is still processing the uploaded file'),
+10 -1
View File
@@ -149,7 +149,16 @@ export const docsyncLinkInputSchema = z.object({
});
export type DocsyncLinkInput = z.infer<typeof docsyncLinkInputSchema>;
export const docsyncLinkUpdateSchema = docsyncLinkInputSchema.omit({ connectionId: true, scopeKey: true }).partial();
/**
* Which folder or tag a binding points at is fixed once it exists: every
* pairing in `document_sync_items` was made against that container, and moving
* it would leave them all pointing somewhere else. The root fields are left out
* rather than ignored — they were accepted and silently dropped, so a client
* sending one was told the move had happened.
*/
export const docsyncLinkUpdateSchema = docsyncLinkInputSchema
.omit({ connectionId: true, scopeKey: true, remoteRootId: true, remoteRootPath: true })
.partial();
// ── Sync state ───────────────────────────────────────────────────────────────