Prevent stale sync confirmations from deleting or overwriting newer data (#739)

Co-authored-by: Gregory Schier <gschier1990@gmail.com>
This commit is contained in:
Zhizhen HeandGregory Schier authored and GitHub committed 2026-10-15 09:10:33 -07:00
1 parent 62f76c5c36
commit 277bbe1baa
13 files changed
+686 -46

No files matched your search

+148
View File
@@ -0,0 +1,148 @@
import type { FunctionComponent, ReactElement } from "react";
import { beforeEach, expect, test, vi } from "vite-plus/test";
import type { SyncOp } from "@yaakapp-internal/sync/bindings/gen_sync";
import type { DialogInstance } from "../components/Dialogs";
import { syncWorkspace } from "./commands";
const mocks = vi.hoisted(() => ({
calculateSync: vi.fn(),
applySync: vi.fn(),
showDialog: vi.fn<(dialog: DialogInstance) => void>(),
hideDialog: vi.fn(),
showToast: vi.fn(),
}));
vi.mock("@yaakapp-internal/sync", () => mocks);
vi.mock("@yaakapp-internal/models", () => ({
createWorkspaceModel: vi.fn(),
modelTypeLabel: () => "Request",
}));
vi.mock("@yaakapp-internal/ui", () => ({
Banner: "div",
InlineCode: "code",
Table: "table",
TableBody: "tbody",
TableCell: "td",
TableHead: "thead",
TableHeaderCell: "th",
TableRow: "tr",
TruncatedWideTableCell: "td",
}));
vi.mock("../components/core/Button", () => ({ Button: "button" }));
vi.mock("../hooks/useActiveWorkspace", () => ({ activeWorkspaceIdAtom: null }));
vi.mock("../lib/dialog", () => mocks);
vi.mock("../lib/jotai", () => ({ jotaiStore: { get: vi.fn() } }));
vi.mock("../lib/prompt", () => ({ showPrompt: vi.fn() }));
vi.mock("../lib/resolvedModelName", () => ({ resolvedModelNameWithFolders: () => "Request" }));
vi.mock("../lib/toast", () => mocks);
const staleToast = expect.objectContaining({
message: "Files changed while syncing. Review the changes again.",
});
const params = { workspaceId: "wk_test", syncDir: "/sync" };
const deletion = {
type: "dbDelete",
model: { id: "rq_test", model: "http_request" },
state: {},
} as Extract<SyncOp, { type: "dbDelete" }>;
const update = {
type: "dbUpdate",
model: { id: "rq_test", model: "http_request" },
fs: { model: { id: "rq_test", model: "http_request", name: "Updated" } },
state: {},
} as Extract<SyncOp, { type: "dbUpdate" }>;
async function submit(dialog: DialogInstance) {
const render = dialog.render as FunctionComponent<{ hide: () => void }>;
const form = render({ hide: () => mocks.hideDialog(dialog.id) }) as ReactElement<{
onSubmit: (event: { preventDefault: () => void }) => Promise<void>;
}>;
await form.props.onSubmit({ preventDefault: vi.fn() });
}
function latestDialog() {
const dialog = mocks.showDialog.mock.calls.at(-1)?.[0];
if (dialog == null) throw new Error("Expected a sync dialog");
return dialog;
}
beforeEach(() => {
vi.resetAllMocks();
mocks.applySync.mockResolvedValue(true);
});
test("closes an obsolete dialog when the restored file leaves no changes", async () => {
mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([]);
await syncWorkspace.mutateAsync(params);
mocks.applySync.mockResolvedValueOnce(false);
await submit(latestDialog());
expect(mocks.applySync).toHaveBeenCalledTimes(1);
expect(mocks.applySync).toHaveBeenCalledWith(params.workspaceId, params.syncDir, [deletion]);
expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync");
expect(mocks.showDialog).toHaveBeenCalledTimes(1);
});
test("requires another confirmation for a changed plan", async () => {
mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([update]);
await syncWorkspace.mutateAsync(params);
const original = latestDialog();
mocks.applySync.mockResolvedValueOnce(false);
await submit(original);
expect(mocks.applySync).toHaveBeenCalledTimes(1);
expect(mocks.showDialog).toHaveBeenCalledTimes(2);
expect(mocks.showToast).toHaveBeenCalledWith(staleToast);
expect(mocks.hideDialog).not.toHaveBeenCalled();
await submit(latestDialog());
expect(mocks.applySync).toHaveBeenLastCalledWith(params.workspaceId, params.syncDir, [update]);
expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync");
});
test("applies an unchanged reviewed plan and closes the dialog", async () => {
mocks.calculateSync.mockResolvedValue([update]);
await syncWorkspace.mutateAsync(params);
await submit(latestDialog());
expect(mocks.calculateSync).toHaveBeenCalledTimes(1);
expect(mocks.applySync).toHaveBeenCalledTimes(1);
expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync");
});
test("forced sync retries the current plan without requesting another confirmation", async () => {
mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([update]);
mocks.applySync.mockResolvedValueOnce(false);
await syncWorkspace.mutateAsync({ ...params, force: true });
expect(mocks.applySync).toHaveBeenCalledTimes(2);
expect(mocks.applySync).toHaveBeenLastCalledWith(params.workspaceId, params.syncDir, [update]);
expect(mocks.showDialog).not.toHaveBeenCalled();
});
test("outbound-only changes clear the old dialog and sync automatically", async () => {
const outbound = { type: "fsUpdate", model: update.model, state: {} } as SyncOp;
mocks.calculateSync.mockResolvedValueOnce([deletion]).mockResolvedValueOnce([outbound]);
await syncWorkspace.mutateAsync(params);
await syncWorkspace.mutateAsync(params);
expect(mocks.applySync).toHaveBeenCalledWith(params.workspaceId, params.syncDir, [outbound]);
expect(mocks.hideDialog).toHaveBeenCalledWith("commit-sync");
expect(mocks.showDialog).toHaveBeenCalledTimes(1);
});
test("stops retrying a plan that keeps going stale", async () => {
mocks.calculateSync.mockResolvedValue([update]);
mocks.applySync.mockResolvedValue(false);
await syncWorkspace.mutateAsync({ ...params, force: true });
expect(mocks.calculateSync).toHaveBeenCalledTimes(3);
expect(mocks.applySync).toHaveBeenCalledTimes(3);
expect(mocks.showToast).toHaveBeenCalledTimes(1);
expect(mocks.showToast).toHaveBeenCalledWith(staleToast);
expect(mocks.showDialog).not.toHaveBeenCalled();
});
+33 -10
View File
@@ -14,10 +14,11 @@ import {
} from "@yaakapp-internal/ui";
import { activeWorkspaceIdAtom } from "../hooks/useActiveWorkspace";
import { createFastMutation } from "../hooks/useFastMutation";
import { showDialog } from "../lib/dialog";
import { hideDialog, showDialog } from "../lib/dialog";
import { jotaiStore } from "../lib/jotai";
import { showPrompt } from "../lib/prompt";
import { resolvedModelNameWithFolders } from "../lib/resolvedModelName";
import { showToast } from "../lib/toast";
import { pluralizeCount } from "@yaakapp-internal/lib/pluralize";
export const createFolder = createFastMutation<
@@ -52,24 +53,48 @@ export const createFolder = createFastMutation<
},
});
// Files can keep changing under a plan, so stop recalculating after a few stale applies.
const MAX_SYNC_ATTEMPTS = 3;
function showStaleSyncToast() {
showToast({
id: "sync-stale",
color: "warning",
message: "Files changed while syncing. Review the changes again.",
});
}
export const syncWorkspace = createFastMutation<
void,
void,
{ workspaceId: string; syncDir: string; force?: boolean }
{ workspaceId: string; syncDir: string; force?: boolean; attempt?: number }
>({
mutationKey: [],
mutationFn: async ({ workspaceId, syncDir, force }) => {
mutationFn: async ({ workspaceId, syncDir, force, attempt = 1 }) => {
const ops = (await calculateSync(workspaceId, syncDir)) ?? [];
if (ops.length === 0) {
console.log("Nothing to sync", workspaceId, syncDir);
hideDialog("commit-sync");
return;
}
console.log("Syncing workspace", workspaceId, syncDir, ops);
const dbOps = ops.filter((o) => o.type.startsWith("db"));
if (dbOps.length === 0) {
await applySync(workspaceId, syncDir, ops);
const apply = async () => {
if (await applySync(workspaceId, syncDir, ops)) {
hideDialog("commit-sync");
} else if (attempt < MAX_SYNC_ATTEMPTS) {
await syncWorkspace.mutateAsync({ workspaceId, syncDir, force, attempt: attempt + 1 });
} else {
hideDialog("commit-sync");
showStaleSyncToast();
}
};
if (dbOps.length === 0 || force) {
hideDialog("commit-sync");
await apply();
return;
}
@@ -79,9 +104,8 @@ export const syncWorkspace = createFastMutation<
console.log("Directory changes detected", { dbOps, ops });
if (force) {
await applySync(workspaceId, syncDir, ops);
return;
if (attempt > 1) {
showStaleSyncToast();
}
showDialog({
@@ -93,8 +117,7 @@ export const syncWorkspace = createFastMutation<
className="h-full grid grid-rows-[auto_auto_minmax(0,1fr)_auto] gap-3"
onSubmit={async (e) => {
e.preventDefault();
await applySync(workspaceId, syncDir, ops);
hide();
await apply();
}}
>
{isDeletingWorkspace ? (
@@ -17,7 +17,7 @@ export const openWorkspaceFromSyncDir = createFastMutation<void, void, string>({
return;
}
await applySync(workspace.id, dir, ops);
await applySync(workspace.id, dir, ops, { fsOnly: true });
await router.navigate({
to: "/workspaces/$workspaceId",
+2 -1
View File
@@ -1129,12 +1129,13 @@ async fn cmd_sync_calculate_fs<R: Runtime>(
Ok(crate::sync_ext::cmd_sync_calculate_fs(&req.dir).await?)
}
async fn cmd_sync_apply<R: Runtime>(ctx: ClientCtx<R>, req: CmdSyncApplyReq) -> Result<()> {
async fn cmd_sync_apply<R: Runtime>(ctx: ClientCtx<R>, req: CmdSyncApplyReq) -> Result<bool> {
Ok(crate::sync_ext::cmd_sync_apply(
ctx.window.app_handle().clone(),
req.sync_ops,
&req.sync_dir,
&req.workspace_id,
req.fs_only,
)
.await?)
}
+19 -5
View File
@@ -12,8 +12,8 @@ use tokio::sync::watch;
use yaak_rpc_schema::WatchResult;
use yaak_sync::error::Error::InvalidSyncDirectory;
use yaak_sync::sync::{
FsCandidate, SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_state_ops,
compute_sync_ops, get_db_candidates, get_fs_candidates,
FsCandidate, SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_ops,
apply_sync_state_ops, compute_sync_ops, get_db_candidates, get_fs_candidates,
};
use yaak_sync::watch::{WatchEvent, watch_directory};
@@ -48,14 +48,28 @@ pub(crate) async fn cmd_sync_apply<R: Runtime>(
sync_ops: Vec<SyncOp>,
sync_dir: &Path,
workspace_id: &str,
) -> Result<()> {
fs_only: bool,
) -> Result<bool> {
let blobs = app_handle.blob_manager();
if !fs_only || sync_ops.iter().any(|op| !matches!(op, SyncOp::DbCreate { .. })) {
let version = app_handle.package_info().version.to_string();
return Ok(apply_sync_ops(
&app_handle.db_manager(),
&blobs,
&version,
workspace_id,
sync_dir,
sync_ops,
)?);
}
// Opening a sync directory imports its files without existing DB sync state.
// Files first, so the write transaction never waits on the filesystem
let pending = apply_fs_sync_ops(workspace_id, sync_dir, sync_ops)?;
let blobs = app_handle.blob_manager();
app_handle.db_manager().with_tx(|tx| {
let sync_state_ops = apply_db_sync_ops(tx, &blobs, workspace_id, sync_dir, pending)?;
apply_sync_state_ops(tx, workspace_id, sync_dir, sync_state_ops)?;
Ok(())
Ok(true)
})
}
+2 -2
View File
@@ -197,7 +197,7 @@ export type CmdSendHttpRequestReq = { environmentId: string | null, cookieJarId:
export type CmdSetWorkspaceKeyReq = { workspaceId: string, key: string, };
export type CmdSyncApplyReq = { syncOps: Array<SyncOp>, syncDir: string, workspaceId: string, };
export type CmdSyncApplyReq = { syncOps: Array<SyncOp>, syncDir: string, workspaceId: string, fsOnly: boolean, };
export type CmdSyncCalculateFsReq = { dir: string, };
@@ -256,6 +256,6 @@ export type ModelsWebsocketEventsReq = { connectionId: string, };
export type ModelsWorkspaceModelsReq = { workspaceId: string | null, };
export type RpcSchema = { cmd_metadata: [CmdMetadataReq, AppMetaData], cmd_template_tokens_to_string: [CmdTemplateTokensToStringReq, string], cmd_render_template: [CmdRenderTemplateReq, string], cmd_send_feedback: [CmdSendFeedbackReq, null], cmd_dismiss_notification: [CmdDismissNotificationReq, null], cmd_grpc_reflect: [CmdGrpcReflectReq, Array<ServiceDefinition>], cmd_grpc_go: [CmdGrpcGoReq, string], cmd_restart: [CmdRestartReq, null], cmd_send_ephemeral_request: [CmdSendEphemeralRequestReq, EphemeralHttpResponse], cmd_format_json: [CmdFormatJsonReq, string], cmd_format_graphql: [CmdFormatGraphqlReq, string], cmd_http_response_body: [CmdHttpResponseBodyReq, FilterResponse], cmd_http_response_body_path: [CmdHttpResponseBodyPathReq, string | null], cmd_http_request_body: [CmdHttpRequestBodyReq, Array<number> | null], cmd_get_sse_events: [CmdGetSseEventsReq, Array<ServerSentEvent>], cmd_get_http_response_events: [CmdGetHttpResponseEventsReq, Array<HttpResponseEvent>], cmd_import_data: [CmdImportDataReq, ImportPlan], cmd_commit_import: [CmdCommitImportReq, BatchUpsertResult], cmd_list_import_sources: [CmdListImportSourcesReq, Array<ImportSource>], cmd_import_sources_for_origin: [CmdImportSourcesForOriginReq, Array<ImportSource>], cmd_detect_import_source: [CmdDetectImportSourceReq, string], cmd_http_request_actions: [CmdHttpRequestActionsReq, Array<GetHttpRequestActionsResponse>], cmd_websocket_request_actions: [CmdWebsocketRequestActionsReq, Array<GetWebsocketRequestActionsResponse>], cmd_call_websocket_request_action: [CmdCallWebsocketRequestActionReq, null], cmd_workspace_actions: [CmdWorkspaceActionsReq, Array<GetWorkspaceActionsResponse>], cmd_call_workspace_action: [CmdCallWorkspaceActionReq, null], cmd_folder_actions: [CmdFolderActionsReq, Array<GetFolderActionsResponse>], cmd_call_folder_action: [CmdCallFolderActionReq, null], cmd_grpc_request_actions: [CmdGrpcRequestActionsReq, Array<GetGrpcRequestActionsResponse>], cmd_template_function_summaries: [CmdTemplateFunctionSummariesReq, Array<GetTemplateFunctionSummaryResponse>], cmd_template_function_config: [CmdTemplateFunctionConfigReq, GetTemplateFunctionConfigResponse], cmd_get_http_authentication_summaries: [CmdGetHttpAuthenticationSummariesReq, Array<GetHttpAuthenticationSummaryResponse>], cmd_get_http_authentication_config: [CmdGetHttpAuthenticationConfigReq, GetHttpAuthenticationConfigResponse], cmd_call_http_request_action: [CmdCallHttpRequestActionReq, null], cmd_call_grpc_request_action: [CmdCallGrpcRequestActionReq, null], cmd_call_http_authentication_action: [CmdCallHttpAuthenticationActionReq, null], cmd_curl_to_request: [CmdCurlToRequestReq, HttpRequest], cmd_export_data: [CmdExportDataReq, string], cmd_create_example_workspace: [CmdCreateExampleWorkspaceReq, BatchUpsertResult], cmd_save_base64_to_binary: [CmdSaveBase64ToBinaryReq, null], cmd_save_response: [CmdSaveResponseReq, null], cmd_send_http_request: [CmdSendHttpRequestReq, HttpResponse], cmd_reload_plugins: [CmdReloadPluginsReq, Array<[string, string]>], cmd_plugin_info: [CmdPluginInfoReq, PluginMetadata], cmd_delete_all_grpc_connections: [CmdDeleteAllGrpcConnectionsReq, null], cmd_delete_send_history: [CmdDeleteSendHistoryReq, null], cmd_delete_all_http_responses: [CmdDeleteAllHttpResponsesReq, null], cmd_get_workspace_meta: [CmdGetWorkspaceMetaReq, WorkspaceMeta], cmd_new_child_window: [CmdNewChildWindowReq, null], cmd_new_main_window: [CmdNewMainWindowReq, null], cmd_check_for_updates: [CmdCheckForUpdatesReq, boolean], cmd_decrypt_template: [CmdDecryptTemplateReq, string], cmd_secure_template: [CmdSecureTemplateReq, string], cmd_get_themes: [CmdGetThemesReq, Array<GetThemesResponse>], cmd_enable_encryption: [CmdEnableEncryptionReq, null], cmd_reveal_workspace_key: [CmdRevealWorkspaceKeyReq, string], cmd_set_workspace_key: [CmdSetWorkspaceKeyReq, null], cmd_disable_encryption: [CmdDisableEncryptionReq, null], cmd_default_headers: [CmdDefaultHeadersReq, Array<HttpRequestHeader>], models_upsert: [ModelsUpsertReq, string], models_delete: [ModelsDeleteReq, string], models_duplicate: [ModelsDuplicateReq, string], models_websocket_events: [ModelsWebsocketEventsReq, Array<WebsocketEvent>], models_grpc_events: [ModelsGrpcEventsReq, Array<GrpcEvent>], models_get_settings: [ModelsGetSettingsReq, Settings], models_get_graphql_introspection: [ModelsGetGraphqlIntrospectionReq, GraphQlIntrospection | null], models_upsert_graphql_introspection: [ModelsUpsertGraphqlIntrospectionReq, GraphQlIntrospection], models_workspace_models: [ModelsWorkspaceModelsReq, string], cmd_git_checkout: [CmdGitCheckoutReq, string], cmd_git_branch: [CmdGitBranchReq, null], cmd_git_delete_branch: [CmdGitDeleteBranchReq, BranchDeleteResult], cmd_git_delete_remote_branch: [CmdGitDeleteRemoteBranchReq, null], cmd_git_merge_branch: [CmdGitMergeBranchReq, null], cmd_git_rename_branch: [CmdGitRenameBranchReq, null], cmd_git_status: [CmdGitStatusReq, GitStatusSummary], cmd_git_branch_info: [CmdGitBranchInfoReq, GitBranchInfoLine truncated
export type RpcSchema = { cmd_metadata: [CmdMetadataReq, AppMetaData], cmd_template_tokens_to_string: [CmdTemplateTokensToStringReq, string], cmd_render_template: [CmdRenderTemplateReq, string], cmd_send_feedback: [CmdSendFeedbackReq, null], cmd_dismiss_notification: [CmdDismissNotificationReq, null], cmd_grpc_reflect: [CmdGrpcReflectReq, Array<ServiceDefinition>], cmd_grpc_go: [CmdGrpcGoReq, string], cmd_restart: [CmdRestartReq, null], cmd_send_ephemeral_request: [CmdSendEphemeralRequestReq, EphemeralHttpResponse], cmd_format_json: [CmdFormatJsonReq, string], cmd_format_graphql: [CmdFormatGraphqlReq, string], cmd_http_response_body: [CmdHttpResponseBodyReq, FilterResponse], cmd_http_response_body_path: [CmdHttpResponseBodyPathReq, string | null], cmd_http_request_body: [CmdHttpRequestBodyReq, Array<number> | null], cmd_get_sse_events: [CmdGetSseEventsReq, Array<ServerSentEvent>], cmd_get_http_response_events: [CmdGetHttpResponseEventsReq, Array<HttpResponseEvent>], cmd_import_data: [CmdImportDataReq, ImportPlan], cmd_commit_import: [CmdCommitImportReq, BatchUpsertResult], cmd_list_import_sources: [CmdListImportSourcesReq, Array<ImportSource>], cmd_import_sources_for_origin: [CmdImportSourcesForOriginReq, Array<ImportSource>], cmd_detect_import_source: [CmdDetectImportSourceReq, string], cmd_http_request_actions: [CmdHttpRequestActionsReq, Array<GetHttpRequestActionsResponse>], cmd_websocket_request_actions: [CmdWebsocketRequestActionsReq, Array<GetWebsocketRequestActionsResponse>], cmd_call_websocket_request_action: [CmdCallWebsocketRequestActionReq, null], cmd_workspace_actions: [CmdWorkspaceActionsReq, Array<GetWorkspaceActionsResponse>], cmd_call_workspace_action: [CmdCallWorkspaceActionReq, null], cmd_folder_actions: [CmdFolderActionsReq, Array<GetFolderActionsResponse>], cmd_call_folder_action: [CmdCallFolderActionReq, null], cmd_grpc_request_actions: [CmdGrpcRequestActionsReq, Array<GetGrpcRequestActionsResponse>], cmd_template_function_summaries: [CmdTemplateFunctionSummariesReq, Array<GetTemplateFunctionSummaryResponse>], cmd_template_function_config: [CmdTemplateFunctionConfigReq, GetTemplateFunctionConfigResponse], cmd_get_http_authentication_summaries: [CmdGetHttpAuthenticationSummariesReq, Array<GetHttpAuthenticationSummaryResponse>], cmd_get_http_authentication_config: [CmdGetHttpAuthenticationConfigReq, GetHttpAuthenticationConfigResponse], cmd_call_http_request_action: [CmdCallHttpRequestActionReq, null], cmd_call_grpc_request_action: [CmdCallGrpcRequestActionReq, null], cmd_call_http_authentication_action: [CmdCallHttpAuthenticationActionReq, null], cmd_curl_to_request: [CmdCurlToRequestReq, HttpRequest], cmd_export_data: [CmdExportDataReq, string], cmd_create_example_workspace: [CmdCreateExampleWorkspaceReq, BatchUpsertResult], cmd_save_base64_to_binary: [CmdSaveBase64ToBinaryReq, null], cmd_save_response: [CmdSaveResponseReq, null], cmd_send_http_request: [CmdSendHttpRequestReq, HttpResponse], cmd_reload_plugins: [CmdReloadPluginsReq, Array<[string, string]>], cmd_plugin_info: [CmdPluginInfoReq, PluginMetadata], cmd_delete_all_grpc_connections: [CmdDeleteAllGrpcConnectionsReq, null], cmd_delete_send_history: [CmdDeleteSendHistoryReq, null], cmd_delete_all_http_responses: [CmdDeleteAllHttpResponsesReq, null], cmd_get_workspace_meta: [CmdGetWorkspaceMetaReq, WorkspaceMeta], cmd_new_child_window: [CmdNewChildWindowReq, null], cmd_new_main_window: [CmdNewMainWindowReq, null], cmd_check_for_updates: [CmdCheckForUpdatesReq, boolean], cmd_decrypt_template: [CmdDecryptTemplateReq, string], cmd_secure_template: [CmdSecureTemplateReq, string], cmd_get_themes: [CmdGetThemesReq, Array<GetThemesResponse>], cmd_enable_encryption: [CmdEnableEncryptionReq, null], cmd_reveal_workspace_key: [CmdRevealWorkspaceKeyReq, string], cmd_set_workspace_key: [CmdSetWorkspaceKeyReq, null], cmd_disable_encryption: [CmdDisableEncryptionReq, null], cmd_default_headers: [CmdDefaultHeadersReq, Array<HttpRequestHeader>], models_upsert: [ModelsUpsertReq, string], models_delete: [ModelsDeleteReq, string], models_duplicate: [ModelsDuplicateReq, string], models_websocket_events: [ModelsWebsocketEventsReq, Array<WebsocketEvent>], models_grpc_events: [ModelsGrpcEventsReq, Array<GrpcEvent>], models_get_settings: [ModelsGetSettingsReq, Settings], models_get_graphql_introspection: [ModelsGetGraphqlIntrospectionReq, GraphQlIntrospection | null], models_upsert_graphql_introspection: [ModelsUpsertGraphqlIntrospectionReq, GraphQlIntrospection], models_workspace_models: [ModelsWorkspaceModelsReq, string], cmd_git_checkout: [CmdGitCheckoutReq, string], cmd_git_branch: [CmdGitBranchReq, null], cmd_git_delete_branch: [CmdGitDeleteBranchReq, BranchDeleteResult], cmd_git_delete_remote_branch: [CmdGitDeleteRemoteBranchReq, null], cmd_git_merge_branch: [CmdGitMergeBranchReq, null], cmd_git_rename_branch: [CmdGitRenameBranchReq, null], cmd_git_status: [CmdGitStatusReq, GitStatusSummary], cmd_git_branch_info: [CmdGitBranchInfoReq, GitBranchInfoLine truncated
export type WatchResult = { unlistenEvent: string, };
+4 -1
View File
@@ -3,4 +3,7 @@ import type { SyncModel, SyncState } from "./gen_models";
export type FsCandidate = { "type": "FsCandidate", model: SyncModel, relPath: string, checksum: string, };
export type SyncOp = { "type": "fsCreate", model: SyncModel, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, };
/**
* Outbound operations record the file candidate checksum, or `None` if it was absent.
*/
export type SyncOp = { "type": "fsCreate", model: SyncModel, fsChecksum: string | null, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, fsChecksum: string | null, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", model: SyncModel, state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, };
+3 -1
View File
@@ -834,6 +834,8 @@ pub struct CmdSyncApplyReq {
pub sync_ops: Vec<SyncOp>,
pub sync_dir: PathBuf,
pub workspace_id: String,
#[serde(default)]
pub fs_only: bool,
}
#[derive(Debug, Deserialize, TS)]
@@ -1024,7 +1026,7 @@ macro_rules! with_commands {
cmd_git_rm_remote(CmdGitRmRemoteReq) -> (),
cmd_sync_calculate(CmdSyncCalculateReq) -> Vec<SyncOp>,
cmd_sync_calculate_fs(CmdSyncCalculateFsReq) -> Vec<SyncOp>,
cmd_sync_apply(CmdSyncApplyReq) -> (),
cmd_sync_apply(CmdSyncApplyReq) -> bool,
cmd_ws_delete_connections(CmdWsDeleteConnectionsReq) -> (),
cmd_ws_send(CmdWsSendReq) -> WebsocketConnection,
cmd_ws_close(CmdWsCloseReq) -> WebsocketConnection,
+1 -1
View File
@@ -2839,7 +2839,7 @@ impl UpsertModelInfo for Plugin {
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default, TS)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default, TS)]
#[serde(default, rename_all = "camelCase")]
#[ts(export, export_to = "gen_models.ts")]
#[enum_def(table_name = "sync_states")]
+4 -1
View File
@@ -3,4 +3,7 @@ import type { SyncModel, SyncState } from "./gen_models";
export type FsCandidate = { "type": "FsCandidate", model: SyncModel, relPath: string, checksum: string, };
export type SyncOp = { "type": "fsCreate", model: SyncModel, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, };
/**
* Outbound operations record the file candidate checksum, or `None` if it was absent.
*/
export type SyncOp = { "type": "fsCreate", model: SyncModel, fsChecksum: string | null, } | { "type": "fsUpdate", model: SyncModel, state: SyncState, fsChecksum: string | null, } | { "type": "fsDelete", state: SyncState, fs: FsCandidate | null, } | { "type": "dbCreate", fs: FsCandidate, } | { "type": "dbUpdate", model: SyncModel, state: SyncState, fs: FsCandidate, } | { "type": "dbDelete", model: SyncModel, state: SyncState, } | { "type": "ignorePrivate", model: SyncModel, };
+8 -2
View File
@@ -16,11 +16,17 @@ export async function calculateSyncFsOnly(dir: string) {
return platform.rpc<SyncOp[]>("cmd_sync_calculate_fs", { dir });
}
export async function applySync(workspaceId: string, syncDir: string, syncOps: SyncOp[]) {
return platform.rpc<void>("cmd_sync_apply", {
export async function applySync(
workspaceId: string,
syncDir: string,
syncOps: SyncOp[],
options?: { fsOnly: boolean },
) {
return platform.rpc<boolean>("cmd_sync_apply", {
workspaceId,
syncDir,
syncOps: syncOps,
fsOnly: options?.fsOnly ?? false,
});
}
+138 -17
View File
@@ -17,18 +17,24 @@ use yaak_models::models::{
Environment, Folder, GrpcRequest, HttpRequest, SyncState, WebsocketRequest, Workspace,
WorkspaceMeta,
};
use yaak_models::query_manager::QueryManager;
use yaak_models::util::{UpdateSource, get_workspace_export_resources};
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
/// Outbound operations record the file candidate checksum, or `None` if it was absent.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase", tag = "type")]
#[ts(export, export_to = "gen_sync.ts")]
pub enum SyncOp {
FsCreate {
model: SyncModel,
#[serde(rename = "fsChecksum")]
fs_checksum: Option<String>,
},
FsUpdate {
model: SyncModel,
state: SyncState,
#[serde(rename = "fsChecksum")]
fs_checksum: Option<String>,
},
FsDelete {
state: SyncState,
@@ -38,6 +44,8 @@ pub enum SyncOp {
fs: FsCandidate,
},
DbUpdate {
// The DB version the user reviewed, so a subsequent edit invalidates the plan.
model: SyncModel,
state: SyncState,
fs: FsCandidate,
},
@@ -56,7 +64,7 @@ impl SyncOp {
SyncOp::DbCreate { fs } => fs.model.workspace_id(),
SyncOp::DbDelete { model, .. } => model.workspace_id(),
SyncOp::DbUpdate { state, .. } => state.workspace_id.clone(),
SyncOp::FsCreate { model } => model.workspace_id(),
SyncOp::FsCreate { model, .. } => model.workspace_id(),
SyncOp::FsDelete { state, .. } => state.workspace_id.clone(),
SyncOp::FsUpdate { state, .. } => state.workspace_id.clone(),
SyncOp::IgnorePrivate { model } => model.workspace_id(),
@@ -68,7 +76,7 @@ impl Display for SyncOp {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
f.write_str(
match self {
SyncOp::FsCreate { model } => format!("fs_create({})", model.id()),
SyncOp::FsCreate { model, .. } => format!("fs_create({})", model.id()),
SyncOp::FsUpdate { model, .. } => format!("fs_update({})", model.id()),
SyncOp::FsDelete { state, .. } => format!("fs_delete({})", state.model_id),
SyncOp::DbCreate { fs } => format!("db_create({})", fs.model.id()),
@@ -101,7 +109,7 @@ impl DbCandidate {
}
}
#[derive(Debug, Clone, Serialize, Deserialize, TS)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase", tag = "type")]
#[ts(export, export_to = "gen_sync.ts")]
pub struct FsCandidate {
@@ -285,13 +293,15 @@ pub fn compute_sync_ops(
}
// DB modified <-> FS missing
(Some(DbCandidate::Modified(model, sync_state)), None) => {
SyncOp::FsUpdate { model: model.to_owned(), state: sync_state.to_owned() }
}
(Some(DbCandidate::Modified(model, sync_state)), None) => SyncOp::FsUpdate {
model: model.to_owned(),
state: sync_state.to_owned(),
fs_checksum: None,
},
// DB added <-> FS missing
(Some(DbCandidate::Added(model)), None) => {
SyncOp::FsCreate { model: model.to_owned() }
SyncOp::FsCreate { model: model.to_owned(), fs_checksum: None }
}
// DB deleted <-> FS missing
@@ -301,11 +311,12 @@ pub fn compute_sync_ops(
}
// DB unchanged <-> FS exists
(Some(DbCandidate::Unmodified(_, sync_state)), Some(fs_candidate)) => {
(Some(DbCandidate::Unmodified(model, sync_state)), Some(fs_candidate)) => {
if sync_state.checksum == fs_candidate.checksum {
return None;
} else {
SyncOp::DbUpdate {
model: model.to_owned(),
state: sync_state.to_owned(),
fs: fs_candidate.to_owned(),
}
@@ -315,23 +326,35 @@ pub fn compute_sync_ops(
// DB modified <-> FS exists
(Some(DbCandidate::Modified(model, sync_state)), Some(fs_candidate)) => {
if sync_state.checksum == fs_candidate.checksum {
SyncOp::FsUpdate { model: model.to_owned(), state: sync_state.to_owned() }
SyncOp::FsUpdate {
model: model.to_owned(),
state: sync_state.to_owned(),
fs_checksum: Some(fs_candidate.checksum.clone()),
}
} else if model.updated_at() < fs_candidate.model.updated_at() {
// CONFLICT! Write to DB if the fs model is newer
SyncOp::DbUpdate {
model: model.to_owned(),
state: sync_state.to_owned(),
fs: fs_candidate.to_owned(),
}
} else {
// CONFLICT! Write to FS if the db model is newer
SyncOp::FsUpdate { model: model.to_owned(), state: sync_state.to_owned() }
SyncOp::FsUpdate {
model: model.to_owned(),
state: sync_state.to_owned(),
fs_checksum: Some(fs_candidate.checksum.clone()),
}
}
}
// DB added <-> FS anything
(Some(DbCandidate::Added(model)), Some(_)) => {
(Some(DbCandidate::Added(model)), Some(fs_candidate)) => {
// This would be super rare (impossible?), so let's follow the user's intention
SyncOp::FsCreate { model: model.to_owned() }
SyncOp::FsCreate {
model: model.to_owned(),
fs_checksum: Some(fs_candidate.checksum.clone()),
}
}
// DB deleted <-> FS exists
@@ -384,6 +407,104 @@ fn workspace_models(db: &ClientDb, version: &str, workspace_id: &str) -> Result<
Ok(sync_models)
}
/// Apply a reviewed workspace plan only while its inputs still match.
/// Returns `false` when the caller needs to calculate and review a new plan.
pub fn apply_sync_ops(
db: &QueryManager,
blobs: &BlobManager,
version: &str,
workspace_id: &str,
sync_dir: &Path,
mut sync_ops: Vec<SyncOp>,
) -> Result<bool> {
let db_candidates = get_db_candidates(&db.connect(), version, workspace_id, sync_dir)?;
let fs_candidates = workspace_fs_candidates(sync_dir, workspace_id, &db_candidates)?;
if !sync_ops_match(&sync_ops, compute_sync_ops(db_candidates, fs_candidates.clone()), false) {
return Ok(false);
}
let pending = apply_fs_sync_ops(workspace_id, sync_dir, sync_ops.clone())?;
let has_db_ops = sync_ops.iter().any(is_db_op);
// File deletions are complete; their remaining work only removes sync state.
sync_ops.retain(|op| !matches!(op, SyncOp::FsDelete { .. }));
// Inbound changes need a fresh file scan. It happens here, before the write
// transaction, so the transaction never waits on the filesystem.
let fs_candidates = if has_db_ops {
let db_candidates = get_db_candidates(&db.connect(), version, workspace_id, sync_dir)?;
let fs_candidates = workspace_fs_candidates(sync_dir, workspace_id, &db_candidates)?;
if !written_sync_ops_match(&sync_ops, db_candidates, fs_candidates.clone()) {
return Ok(false);
}
fs_candidates
} else {
fs_candidates
};
db.with_tx(|tx| {
// Another writer may have changed the DB while files were being written or
// while waiting for the writer. Files were checked above; only the DB is reread.
let db_candidates = get_db_candidates(tx, version, workspace_id, sync_dir)?;
if !written_sync_ops_match(&sync_ops, db_candidates, fs_candidates) {
return Ok(false);
}
let states = apply_db_sync_ops(tx, blobs, workspace_id, sync_dir, pending)?;
apply_sync_state_ops(tx, workspace_id, sync_dir, states)?;
Ok(true)
})
}
/// The sync directory's candidates for one workspace.
fn workspace_fs_candidates(
sync_dir: &Path,
workspace_id: &str,
db_candidates: &[DbCandidate],
) -> Result<Vec<FsCandidate>> {
if !sync_dir.exists() {
return Err(Error::InvalidSyncDirectory(sync_dir.to_string_lossy().to_string()));
}
Ok(get_fs_candidates(sync_dir, db_candidates)?
.into_iter()
.filter(|fs| fs.model.workspace_id() == workspace_id)
.collect())
}
/// Whether a plan, minus its completed file deletions, still matches once its files are written.
fn written_sync_ops_match(
expected: &[SyncOp],
db_candidates: Vec<DbCandidate>,
fs_candidates: Vec<FsCandidate>,
) -> bool {
let current = compute_sync_ops(db_candidates, fs_candidates)
.into_iter()
.filter(|op| !matches!(op, SyncOp::FsDelete { .. }))
.collect();
sync_ops_match(expected, current, true)
}
fn sync_ops_match(expected: &[SyncOp], current: Vec<SyncOp>, files_written: bool) -> bool {
// The join emits one operation per ID, in arbitrary order.
expected.len() == current.len()
&& current.iter().all(|op| {
expected.iter().any(|reviewed| match (op, reviewed) {
// Once outbound files are written, only their DB/state inputs must still match.
(
SyncOp::FsCreate { model, .. },
SyncOp::FsCreate { model: reviewed_model, .. },
) if files_written => model == reviewed_model,
(
SyncOp::FsUpdate { model, state, .. },
SyncOp::FsUpdate { model: reviewed_model, state: reviewed_state, .. },
) if files_written => model == reviewed_model && state == reviewed_state,
_ => op == reviewed,
})
})
}
fn is_db_op(op: &SyncOp) -> bool {
matches!(op, SyncOp::DbCreate { .. } | SyncOp::DbUpdate { .. } | SyncOp::DbDelete { .. })
}
/// The database half of a sync apply, ready to run once the files are on disk.
pub struct PendingDbSyncOps {
sync_state_ops: Vec<SyncStateOp>,
@@ -399,7 +520,7 @@ pub struct PendingDbSyncOps {
/// Apply the filesystem half of the sync operations: create, rewrite and
/// delete files. Returns the database half, for [`apply_db_sync_ops`].
///
/// Split this way so the file work, which can be slow, happens before the
/// Split this way so filesystem writes, which can be slow, happen before the
/// write transaction is opened rather than inside it.
pub fn apply_fs_sync_ops(
workspace_id: &str,
@@ -432,14 +553,14 @@ pub fn apply_fs_sync_ops(
}
let state_op = match op {
SyncOp::FsCreate { model } => {
SyncOp::FsCreate { model, .. } => {
let rel_path = derive_model_filename(&model);
let abs_path = sync_dir.join(rel_path.clone());
let (content, checksum) = model.to_file_contents(&rel_path)?;
write_sync_file(&abs_path, |file| file.write_all(&content))?;
SyncStateOp::Create { model_id: model.id(), checksum, rel_path }
}
SyncOp::FsUpdate { model, state } => {
SyncOp::FsUpdate { model, state, .. } => {
// Always write the existing path
let rel_path = Path::new(&state.rel_path);
let abs_path = Path::new(&state.sync_dir).join(&rel_path);
@@ -470,7 +591,7 @@ pub fn apply_fs_sync_ops(
rel_path: fs.rel_path.to_owned(),
}
}
SyncOp::DbUpdate { state, fs } => {
SyncOp::DbUpdate { state, fs, .. } => {
pending.push_upsert(fs.model);
SyncStateOp::Update {
state: state.to_owned(),
+323 -4
View File
@@ -2,13 +2,14 @@ use std::fs;
use std::path::PathBuf;
use tempfile::TempDir;
use yaak_models::blob_manager::BlobManager;
use yaak_models::models::{HttpRequest, SyncState, Workspace};
use yaak_models::models::{HttpRequest, HttpResponse, SyncState, Workspace};
use yaak_models::query_manager::QueryManager;
use yaak_models::util::UpdateSource;
use yaak_sync::error::{Error, Result};
use yaak_sync::models::SyncModel;
use yaak_sync::sync::{
SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_state_ops, compute_sync_ops,
get_db_candidates, get_fs_candidates,
SyncOp, apply_db_sync_ops, apply_fs_sync_ops, apply_sync_ops, apply_sync_state_ops,
compute_sync_ops, get_db_candidates, get_fs_candidates,
};
struct Fixture {
@@ -22,8 +23,12 @@ struct Fixture {
impl Fixture {
fn new(request_filename: &str) -> Self {
let (db, blobs, _events) = yaak_models::init_in_memory().unwrap();
let dir = tempfile::tempdir().unwrap();
let (db, blobs, _events) = yaak_models::init_standalone(
dir.path().join(".db/models.sqlite"),
dir.path().join(".db/blobs.sqlite"),
)
.unwrap();
let workspace = Workspace { id: "wk_sync_test".into(), ..Default::default() };
let request = HttpRequest {
id: "rq_sync_test".into(),
@@ -75,6 +80,11 @@ impl Fixture {
.unwrap();
}
fn apply_reviewed(&self, ops: Vec<SyncOp>) -> bool {
apply_sync_ops(&self.db, &self.blobs, "test", &self.workspace_id, self.dir.path(), ops)
.unwrap()
}
fn states(&self) -> Vec<SyncState> {
let mut states = self
.db
@@ -215,3 +225,312 @@ fn unrelated_files_are_still_ignored() {
fs::create_dir(fixture.dir.path().join("other-files")).unwrap();
assert!(fixture.calculate().unwrap().is_empty());
}
#[test]
fn restored_file_invalidates_reviewed_deletion_and_preserves_history() {
let fixture = Fixture::new("request.yaml");
let response = fixture
.db
.with_tx::<_, Error>(|tx| {
Ok(tx.upsert_http_response(
&HttpResponse {
id: "rs_sync_test".into(),
request_id: fixture.request.id.clone(),
workspace_id: fixture.workspace_id.clone(),
status: 200,
..Default::default()
},
&UpdateSource::Background,
&fixture.blobs,
)?)
})
.unwrap();
let saved = fs::read(&fixture.request_path).unwrap();
fs::remove_file(&fixture.request_path).unwrap();
let ops = fixture.calculate().unwrap();
assert!(matches!(ops.as_slice(), [SyncOp::DbDelete { .. }]));
fs::write(&fixture.request_path, &saved).unwrap();
let before = fixture.states();
assert!(!fixture.apply_reviewed(ops));
assert_eq!(fixture.db_request().name, "Keep this request");
assert_eq!(
serde_json::to_value(fixture.db.connect().get_http_response(&response.id).unwrap())
.unwrap(),
serde_json::to_value(response).unwrap()
);
assert_eq!(fixture.states(), before);
assert_eq!(fs::read(&fixture.request_path).unwrap(), saved);
assert!(fixture.calculate().unwrap().is_empty());
}
#[test]
fn newer_database_edit_invalidates_reviewed_update_even_when_file_timestamp_wins() {
for file_is_newer in [false, true] {
let fixture = Fixture::new("request.yaml");
fixture.edit(|value| {
value["name"] = "Previously reviewed file version".into();
if file_is_newer {
value["updatedAt"] = serde_json::to_value(
(chrono::Utc::now() + chrono::Duration::days(1)).naive_utc(),
)
.unwrap();
}
});
let ops = fixture.calculate().unwrap();
assert!(matches!(ops.as_slice(), [SyncOp::DbUpdate { .. }]));
let before = fixture.states();
let mut request = fixture.db_request();
request.name = "New edit from another window".into();
fixture
.db
.with_tx::<_, Error>(|tx| {
tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?;
Ok(())
})
.unwrap();
if file_is_newer {
assert!(matches!(fixture.calculate().unwrap().as_slice(), [SyncOp::DbUpdate { .. }]));
}
assert!(!fixture.apply_reviewed(ops));
assert_eq!(fixture.db_request().name, request.name);
assert_eq!(fixture.states(), before);
}
}
#[test]
fn changed_file_invalidates_reviewed_update_until_reviewed_again() {
let fixture = Fixture::new("request.yaml");
fixture.edit(|value| value["name"] = "Reviewed version".into());
let ops = fixture.calculate().unwrap();
fixture.edit(|value| value["name"] = "New file version".into());
let before = fixture.states();
assert!(!fixture.apply_reviewed(ops));
assert_eq!(fixture.db_request().name, "Keep this request");
assert_eq!(fixture.states(), before);
assert!(fixture.apply_reviewed(fixture.calculate().unwrap()));
assert_eq!(fixture.db_request().name, "New file version");
assert!(fixture.calculate().unwrap().is_empty());
}
#[test]
fn unchanged_reviewed_plan_applies_in_any_order_and_cannot_be_replayed() {
let fixture = Fixture::new("request.yaml");
fs::remove_file(&fixture.request_path).unwrap();
let path = fixture.dir.path().join("workspace.yaml");
let mut workspace = fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap();
workspace.name = "Reviewed workspace".into();
let (contents, _) = SyncModel::Workspace(workspace).to_file_contents(&path).unwrap();
fs::write(path, contents).unwrap();
let mut ops = fixture.calculate().unwrap();
assert_eq!(ops.len(), 2);
ops.reverse();
assert!(fixture.apply_reviewed(ops.clone()));
assert!(fixture.db.connect().get_http_request(&fixture.request.id).is_err());
assert_eq!(
fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap().name,
"Reviewed workspace"
);
assert!(fixture.calculate().unwrap().is_empty());
let before = fixture.states();
assert!(!fixture.apply_reviewed(ops));
assert_eq!(fixture.states(), before);
}
#[test]
fn database_changes_while_waiting_for_writer_invalidate_reviewed_plan() {
use std::sync::mpsc;
use std::time::{Duration, Instant};
// Files are checked before the write transaction opens, so only database changes made
// while waiting for the writer are caught here.
let fixture = Fixture::new("request.yaml");
fixture.edit(|value| value["name"] = "Incoming file version".into());
fixture
.db
.with_tx::<_, Error>(|tx| {
let mut workspace = tx.get_workspace(&fixture.workspace_id)?;
workspace.name = "Outbound workspace version".into();
tx.upsert_workspace(&workspace, &UpdateSource::from_window_label("test"))?;
Ok(())
})
.unwrap();
let ops = fixture.calculate().unwrap();
assert_eq!(ops.len(), 2);
let before = fixture.states();
let workspace_path = fixture.dir.path().join("workspace.yaml");
let (ready_tx, ready_rx) = mpsc::channel();
let (resume_tx, resume_rx) = mpsc::channel();
std::thread::scope(|scope| {
let fixture = &fixture;
let writer = scope.spawn(move || {
fixture
.db
.with_tx::<_, Error>(|tx| {
ready_tx.send(()).unwrap();
resume_rx.recv_timeout(Duration::from_secs(10)).unwrap();
let mut request = tx.get_http_request(&fixture.request.id)?;
request.name = "Concurrent database edit".into();
tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?;
Ok(())
})
.unwrap();
});
ready_rx.recv_timeout(Duration::from_secs(5)).unwrap();
let applying = scope.spawn(move || fixture.apply_reviewed(ops));
let deadline = Instant::now() + Duration::from_secs(5);
let mut published = false;
while Instant::now() < deadline {
if fs::read_to_string(&workspace_path).unwrap().contains("Outbound workspace version") {
published = true;
break;
}
std::thread::sleep(Duration::from_millis(10));
}
resume_tx.send(()).unwrap();
writer.join().unwrap();
assert!(!applying.join().unwrap());
assert!(published, "filesystem writes should proceed while the DB writer is held");
});
assert_eq!(fixture.db_request().name, "Concurrent database edit");
assert_eq!(fixture.states(), before);
}
#[test]
fn current_outbound_only_plan_still_syncs_automatically() {
let fixture = Fixture::new("request.yaml");
fixture
.db
.with_tx::<_, Error>(|tx| {
let mut request = tx.get_http_request(&fixture.request.id)?;
request.name = "New database version".into();
tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?;
Ok(())
})
.unwrap();
let ops = fixture.calculate().unwrap();
assert!(matches!(ops.as_slice(), [SyncOp::FsUpdate { .. }]));
assert!(fixture.apply_reviewed(ops));
assert!(fs::read_to_string(&fixture.request_path).unwrap().contains("New database version"));
assert!(fixture.calculate().unwrap().is_empty());
}
#[test]
fn reviewed_import_and_file_deletion_apply_together() {
let fixture = Fixture::new("request.yaml");
let mut added = fixture.request.clone();
added.id = "rq_sync_added".into();
let path = fixture.dir.path().join("added.yaml");
let (contents, _) = SyncModel::HttpRequest(added.clone()).to_file_contents(&path).unwrap();
fs::write(path, contents).unwrap();
fixture
.db
.with_tx::<_, Error>(|tx| {
let request = tx.get_http_request(&fixture.request.id)?;
tx.delete_http_request(&request, &UpdateSource::Sync)?;
Ok(())
})
.unwrap();
let ops = fixture.calculate().unwrap();
assert!(ops.iter().any(|op| matches!(op, SyncOp::DbCreate { .. })));
assert!(ops.iter().any(|op| matches!(op, SyncOp::FsDelete { fs: Some(_), .. })));
assert!(fixture.apply_reviewed(ops));
assert!(!fixture.request_path.exists());
assert_eq!(fixture.db.connect().get_http_request(&added.id).unwrap().id, added.id);
assert!(fixture.calculate().unwrap().is_empty());
}
#[test]
fn reviewed_inbound_and_outbound_updates_apply_together() {
let fixture = Fixture::new("request.yaml");
fixture.edit(|value| value["name"] = "Incoming request".into());
fixture
.db
.with_tx::<_, Error>(|tx| {
let mut workspace = tx.get_workspace(&fixture.workspace_id)?;
workspace.name = "Outgoing workspace".into();
tx.upsert_workspace(&workspace, &UpdateSource::from_window_label("test"))?;
Ok(())
})
.unwrap();
let ops = fixture.calculate().unwrap();
assert!(ops.iter().any(|op| matches!(op, SyncOp::DbUpdate { .. })));
assert!(ops.iter().any(|op| matches!(op, SyncOp::FsUpdate { .. })));
assert!(fixture.apply_reviewed(ops));
assert_eq!(fixture.db_request().name, "Incoming request");
assert!(
fs::read_to_string(fixture.dir.path().join("workspace.yaml"))
.unwrap()
.contains("Outgoing workspace")
);
assert!(fixture.calculate().unwrap().is_empty());
}
#[test]
fn later_file_edits_invalidate_reviewed_outbound_operations() {
for creating in [false, true] {
for file_present in [true, false] {
let fixture = Fixture::new("yaak.rq_sync_test.yaml");
let state = fixture
.states()
.into_iter()
.find(|state| state.model_id == fixture.request.id)
.unwrap();
fixture
.db
.with_tx::<_, Error>(|tx| {
if creating {
tx.delete_sync_state(&state)?;
}
let mut request = tx.get_http_request(&fixture.request.id)?;
request.name = "Reviewed outgoing request".into();
tx.upsert_http_request(&request, &UpdateSource::from_window_label("test"))?;
Ok(())
})
.unwrap();
let workspace_path = fixture.dir.path().join("workspace.yaml");
let workspace = fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap();
let mut incoming = workspace.clone();
incoming.name = "Reviewed incoming workspace".into();
let (content, _) =
SyncModel::Workspace(incoming).to_file_contents(&workspace_path).unwrap();
fs::write(&workspace_path, &content).unwrap();
if !file_present {
fs::remove_file(&fixture.request_path).unwrap();
}
let ops = fixture.calculate().unwrap();
assert!(ops.iter().any(|op| matches!(op, SyncOp::DbUpdate { .. })));
assert!(ops.iter().any(|op| if creating {
matches!(op, SyncOp::FsCreate { .. })
} else {
matches!(op, SyncOp::FsUpdate { .. })
}));
let request = fixture.db_request();
let states = fixture.states();
// Keep updatedAt older than the DB version, so conflict resolution still writes to FS.
fixture.edit(|value| value["name"] = "Later external file edit".into());
let edited = fs::read(&fixture.request_path).unwrap();
assert!(
!fixture.apply_reviewed(ops),
"creating={creating}, file_present={file_present}"
);
assert_eq!(fs::read(&fixture.request_path).unwrap(), edited);
assert_eq!(fs::read(&workspace_path).unwrap(), content);
assert_eq!(fixture.db_request(), request);
assert_eq!(
fixture.db.connect().get_workspace(&fixture.workspace_id).unwrap(),
workspace
);
assert_eq!(fixture.states(), states);
assert!(fixture.apply_reviewed(fixture.calculate().unwrap()));
assert!(fixture.calculate().unwrap().is_empty());
}
}
}