Good catches
This commit is contained in:
parent
0ab6984cdf
commit
debe7cfc37
14 changed files with 201 additions and 63 deletions
|
|
@ -93,6 +93,13 @@ export class CursorTracker {
|
|||
await this.getDocumentsUpToDateness(clientCursor);
|
||||
}
|
||||
}
|
||||
// Drop the local-cursor send-cache so the next call re-reads
|
||||
// the file. The first cache key is the editor's input, which
|
||||
// doesn't change when the file content does — without this,
|
||||
// a remote update flipping the file from dirty back to clean
|
||||
// would never re-send the cursor with a fresh `vaultUpdateId`.
|
||||
this.lastLocalCursorStateJson = "";
|
||||
this.lastLocalCursorStateWithoutDirtyDocumentsJson = "";
|
||||
})
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -46,6 +46,38 @@ export class ExpectedFsEvents {
|
|||
this.bump(this.renames, ExpectedFsEvents.renameKey(oldPath, newPath));
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel a previously-registered expectation when the fs op that registered
|
||||
* it failed before any watcher event could fire. Without this, a leaked
|
||||
* expectation silently swallows the next genuine user event at the same
|
||||
* path (or, for renames, the same `oldPath → newPath` pair).
|
||||
*
|
||||
* Floored at zero: if the watcher *did* fire (op partially completed) and
|
||||
* already consumed the entry, the unexpect is a no-op. The fallback is
|
||||
* acceptable — at worst we re-upload a real edit we'd otherwise filter.
|
||||
*/
|
||||
public unexpectCreate(path: RelativePath): void {
|
||||
this.decrement(this.creates, path);
|
||||
}
|
||||
|
||||
public unexpectUpdate(path: RelativePath): void {
|
||||
this.decrement(this.updates, path);
|
||||
}
|
||||
|
||||
public unexpectDelete(path: RelativePath): void {
|
||||
this.decrement(this.deletes, path);
|
||||
}
|
||||
|
||||
public unexpectRename(
|
||||
oldPath: RelativePath,
|
||||
newPath: RelativePath
|
||||
): void {
|
||||
this.decrement(
|
||||
this.renames,
|
||||
ExpectedFsEvents.renameKey(oldPath, newPath)
|
||||
);
|
||||
}
|
||||
|
||||
public matchCreate(path: RelativePath): boolean {
|
||||
return this.consume(this.creates, path);
|
||||
}
|
||||
|
|
@ -95,4 +127,10 @@ export class ExpectedFsEvents {
|
|||
else {map.set(key, count - 1);}
|
||||
return true;
|
||||
}
|
||||
|
||||
private decrement(map: Map<RelativePath, number>, key: RelativePath): void {
|
||||
const count = map.get(key) ?? 0;
|
||||
if (count <= 1) {map.delete(key);}
|
||||
else {map.set(key, count - 1);}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import type { Logger } from "../tracing/logger";
|
|||
import { globsToRegexes } from "../utils/globs-to-regexes";
|
||||
import { CONFLICT_PATH_REGEX } from "./conflict-path";
|
||||
import { removeFromArray } from "../utils/remove-from-array";
|
||||
import { EventListeners } from "../utils/data-structures/event-listeners";
|
||||
import type { DocumentWithPath } from "./types";
|
||||
import {
|
||||
SyncEventType,
|
||||
|
|
@ -17,6 +18,14 @@ import {
|
|||
import { MinCovered } from "../utils/data-structures/min-covered";
|
||||
|
||||
export class SyncEventQueue {
|
||||
// Fires synchronously whenever the events array length changes (push, pop,
|
||||
// remove, bulk-clear). The Syncer mirrors this into its public count
|
||||
// listener; without this hook, listeners only saw deltas at consume time
|
||||
// and missed the "queue grew" / "queue cleared on reset" transitions.
|
||||
public readonly onPendingUpdateCountChanged = new EventListeners<
|
||||
(count: number) => unknown
|
||||
>();
|
||||
|
||||
private readonly _lastSeenUpdateId: MinCovered;
|
||||
|
||||
// Latest state of the filesystem as we know it, excluding
|
||||
|
|
@ -123,6 +132,7 @@ export class SyncEventQueue {
|
|||
|
||||
if (input.type === SyncEventType.RemoteChange) {
|
||||
this.events.push(input);
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -154,6 +164,7 @@ export class SyncEventQueue {
|
|||
originalPath: path,
|
||||
resolvers: Promise.withResolvers()
|
||||
});
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -180,6 +191,7 @@ export class SyncEventQueue {
|
|||
type: SyncEventType.LocalDelete,
|
||||
documentId: (pendingDocumentId ?? documentId)!
|
||||
});
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -219,6 +231,7 @@ export class SyncEventQueue {
|
|||
path,
|
||||
originalPath: path
|
||||
});
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
|
||||
if (needsSave) {
|
||||
await this.save();
|
||||
|
|
@ -226,7 +239,11 @@ export class SyncEventQueue {
|
|||
}
|
||||
|
||||
public async next(): Promise<SyncEvent | undefined> {
|
||||
return this.events.shift();
|
||||
const event = this.events.shift();
|
||||
if (event !== undefined) {
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
}
|
||||
return event;
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -250,7 +267,9 @@ export class SyncEventQueue {
|
|||
* remote-create handler just absorbed).
|
||||
*/
|
||||
public consumeEvent(event: SyncEvent): void {
|
||||
removeFromArray(this.events, event);
|
||||
if (removeFromArray(this.events, event)) {
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -261,7 +280,9 @@ export class SyncEventQueue {
|
|||
event: Extract<SyncEvent, { type: SyncEventType.LocalCreate }>,
|
||||
record: DocumentRecord
|
||||
): Promise<void> {
|
||||
removeFromArray(this.events, event); // in case the create event is still pending
|
||||
if (removeFromArray(this.events, event)) {
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
}
|
||||
await this.setDocument(event.path, record);
|
||||
event.resolvers.resolve(record.documentId);
|
||||
}
|
||||
|
|
@ -376,8 +397,16 @@ export class SyncEventQueue {
|
|||
}
|
||||
|
||||
public clearPending(): void {
|
||||
const hadEvents = this.events.length > 0;
|
||||
this.rejectAllPendingCreates();
|
||||
this.events.length = 0;
|
||||
if (hadEvents) {
|
||||
this.notifyPendingUpdateCountChanged();
|
||||
}
|
||||
}
|
||||
|
||||
private notifyPendingUpdateCountChanged(): void {
|
||||
this.onPendingUpdateCountChanged.trigger(this.events.length);
|
||||
}
|
||||
|
||||
public findLatestCreateForPath(
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ import type { SyncHistory } from "../tracing/sync-history";
|
|||
import {
|
||||
SyncStatus,
|
||||
SyncType,
|
||||
type CommonHistoryEntry
|
||||
type HistoryEntry
|
||||
} from "../tracing/sync-history";
|
||||
import { isBinary } from "../utils/is-binary";
|
||||
import { isFileTypeMergable } from "../utils/is-file-type-mergable";
|
||||
|
|
@ -72,6 +72,12 @@ export class Syncer {
|
|||
this.webSocketManager.onRemoteVaultUpdateReceived.add(
|
||||
this.syncRemotelyUpdatedFile.bind(this)
|
||||
);
|
||||
// Funnel every queue mutation (enqueue, consume, clearPending) through
|
||||
// the public count notifier so listeners see grow/shrink transitions
|
||||
// immediately rather than only when a drain consumes an event.
|
||||
this.queue.onPendingUpdateCountChanged.add(() => {
|
||||
this.notifyRemainingOperationsChanged();
|
||||
});
|
||||
}
|
||||
|
||||
public syncLocallyCreatedFile(relativePath: RelativePath): void {
|
||||
|
|
@ -152,9 +158,24 @@ export class Syncer {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* True while there is queued or in-flight work the syncer needs to handle:
|
||||
* a running offline scan, an active drain, or pending events. Used by
|
||||
* `SyncClient.waitUntilFinishedInternal` to detect WebSocket-fed work that
|
||||
* landed in the queue after the syncer's first quiescence point.
|
||||
*/
|
||||
public get hasPendingWork(): boolean {
|
||||
return (
|
||||
this.runningScheduleSyncForOfflineChanges !== undefined ||
|
||||
this.drainPromise !== undefined ||
|
||||
this.queue.pendingUpdateCount > 0
|
||||
);
|
||||
}
|
||||
|
||||
public reset(): void {
|
||||
this.queue.clearPending();
|
||||
this.clearOfflineScanGate();
|
||||
this.previousRemainingOperationsCount = 0;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -350,13 +371,19 @@ export class Syncer {
|
|||
event.resolvers.reject(new Error("Create was cancelled"));
|
||||
}
|
||||
|
||||
// Advance the cursor so the server doesn't replay this update on every
|
||||
// reconnect — the skip is permanent for this version.
|
||||
if (event.type === SyncEventType.RemoteChange) {
|
||||
this.queue.lastSeenUpdateId = event.remoteVersion.vaultUpdateId;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private getHistoryEntryForSkippedOversizedFile(
|
||||
sizeInBytes: number,
|
||||
relativePath: RelativePath
|
||||
): CommonHistoryEntry | undefined {
|
||||
): HistoryEntry | undefined {
|
||||
const sizeInMB = Math.round(sizeInBytes / 1024 / 1024);
|
||||
const { maxFileSizeMB } = this.settings.getSettings();
|
||||
if (sizeInMB > maxFileSizeMB) {
|
||||
|
|
@ -366,7 +393,8 @@ export class Syncer {
|
|||
type: SyncType.SKIPPED as const,
|
||||
relativePath
|
||||
},
|
||||
message: `File size of ${sizeInMB} MB exceeds the maximum file size limit of ${maxFileSizeMB} MB`
|
||||
message: `File size of ${sizeInMB} MB exceeds the maximum file size limit of ${maxFileSizeMB} MB`,
|
||||
timestamp: new Date()
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -429,7 +457,6 @@ export class Syncer {
|
|||
// and history entry. Keeping the entry in the map until then lets
|
||||
// late remote updates be recognised as "file is missing" and
|
||||
// skipped, instead of resurrecting the doc.
|
||||
//
|
||||
this.history.addHistoryEntry({
|
||||
status: SyncStatus.SUCCESS,
|
||||
details: {
|
||||
|
|
@ -437,7 +464,8 @@ export class Syncer {
|
|||
relativePath: doc.path
|
||||
},
|
||||
message: "Successfully deleted file on the server",
|
||||
author: response.userId
|
||||
author: response.userId,
|
||||
timestamp: new Date(response.updatedDate)
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -482,8 +510,6 @@ export class Syncer {
|
|||
return;
|
||||
}
|
||||
|
||||
this.queue.lastSeenUpdateId = response.vaultUpdateId;
|
||||
|
||||
await this.handleMaybeMergingResponse({
|
||||
path: diskPath,
|
||||
response,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue