TapTap/FastSync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Merge branch 'feat/parity-delete' into feat/rsync-parity

Browse Source
This commit is contained in:
TapTap
2026-09-15 23:24:24 +02:00
parent 17b0632098 376e6500ab
commit 58b3a33e82
28 changed files with 1216 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/receiver.c
+32 -14
View File
@@ -43,17 +43,19 @@ void receiver_outcomes_destroy(ReceiverOutcomes* outcomes) {
/* End-of-transfer success frame. When --remove-source-files was negotiated
each processed data file is acknowledged first (STATUS_NEXT = written,
STATUS_OK = skipped) so the sender never removes a source the receiver did
not actually store. The frame always ends with a plain STATUS_OK. */
bool receiver_send_final_success(int fd, const Config* config, const ReceiverOutcomes* outcomes) {
not actually store. The frame ends with `final_status` (STATUS_OK, or
STATUS_DELETE_LIMIT when a --max-delete commit was capped). */
bool receiver_send_final_success(int fd, const Config* config, const ReceiverOutcomes* outcomes,
Status final_status) {
if (!config->remove_source_files)
return send_status(fd, STATUS_OK);
return send_status(fd, final_status);
size_t count = outcomes ? outcomes->count : 0;
for (size_t i = 0; i < count; i++) {
Status per_file = outcomes->entries[i] == FILE_SAVE_WRITTEN ? STATUS_NEXT : STATUS_OK;
if (!send_status(fd, per_file))
return false;
}
return send_status(fd, STATUS_OK);
return send_status(fd, final_status);
}
static bool receiver_process_chunk(Chunk* chunk, const ReceiverSink* sink) {
@@ -346,15 +348,19 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
moment it arrives, before any file data. Delete now and acknowledge
so the sender only starts streaming once the deletion committed (or
failed). This is the rsync delete-before/delete-during window: a
later transfer failure does not restore these deletions. */
bool deletion_ok = (config->use_delete || config->delete_missing_args)
? manifest_delete_all(config, manifest)
: true;
later transfer failure does not restore these deletions. A
--max-delete-capped commit still succeeds and the transfer proceeds;
the terminal success frame reports the cap. */
DeleteCommitResult deletion = (config->use_delete || config->delete_missing_args)
? manifest_delete_all(config, manifest)
: DELETE_COMMIT_OK;
delete_manifest_free(manifest);
if (!deletion_ok) {
if (deletion == DELETE_COMMIT_ERROR) {
send_status(file_descriptor, STATUS_ERROR);
goto fail;
}
if (deletion == DELETE_COMMIT_LIMIT_REACHED && sink->note_delete_limit)
sink->note_delete_limit(sink->context);
if (!send_status(file_descriptor, STATUS_OK))
goto fail;
} else if (config->use_delete || config->delete_missing_args) {
@@ -407,13 +413,15 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
*pending_manifest = deferred_manifest;
deferred_manifest = NULL;
} else {
bool deletion_ok = manifest_delete_all(config, deferred_manifest);
DeleteCommitResult deletion = manifest_delete_all(config, deferred_manifest);
delete_manifest_free(deferred_manifest);
deferred_manifest = NULL;
if (!deletion_ok) {
if (deletion == DELETE_COMMIT_ERROR) {
send_status(file_descriptor, STATUS_ERROR);
goto fail;
}
if (deletion == DELETE_COMMIT_LIMIT_REACHED && sink->note_delete_limit)
sink->note_delete_limit(sink->context);
}
}
if (sink->send_success) {
@@ -454,6 +462,9 @@ typedef struct {
after the whole transfer (and its delete/publication phases) has run so a
child write never clobbers a directory mtime. */
DirTimeList dir_times;
/* Set when a --max-delete commit was capped; the terminal frame then carries
STATUS_DELETE_LIMIT so the sender exits 25 like rsync. */
bool delete_limit_reached;
} ReceiverSaveContext;
static bool receiver_save_file(File* file, void* context_pointer) {
@@ -494,12 +505,18 @@ static bool receiver_save_file(File* file, void* context_pointer) {
return result != FILE_SAVE_ERROR;
}
static void receiver_note_delete_limit(void* context_pointer) {
ReceiverSaveContext* context = context_pointer;
context->delete_limit_reached = true;
}
static bool receiver_send_success_frame(int fd, void* context_pointer) {
ReceiverSaveContext* context = context_pointer;
Status final_status = context->delete_limit_reached ? STATUS_DELETE_LIMIT : STATUS_OK;
/* Server-contacting --dry-run: nothing was staged or written, so there is
nothing to publish and no directory times to stamp. */
if (context->config->dry_run)
return receiver_send_final_success(fd, context->config, &context->outcomes);
return receiver_send_final_success(fd, context->config, &context->outcomes, final_status);
/* --delay-updates: the whole protocol stream (including manifest/delete
handling, which ran inside receiver_process) has succeeded and every
staged file was fully written. Publish them atomically now, before the
@@ -517,13 +534,14 @@ static bool receiver_send_success_frame(int fd, void* context_pointer) {
before calling this success frame. */
dir_metadata_list_apply(&context->dir_times, context->config->receive_root_directory,
context->config);
return receiver_send_final_success(fd, context->config, &context->outcomes);
return receiver_send_final_success(fd, context->config, &context->outcomes, final_status);
}
int receiver_receive_files(Config* config, int file_descriptor) {
ReceiverSaveContext context = {.config = config, .outcomes = {0}};
dir_time_list_init(&context.dir_times);
ReceiverSink sink = {receiver_save_file, &context, true, true, receiver_send_success_frame};
ReceiverSink sink = {receiver_save_file, &context, true, true, receiver_send_success_frame,
receiver_note_delete_limit};
int ret = receiver_process(config, file_descriptor, &sink);
if (ret != 0 && config->delay_updates && config->delay_context)
delay_updates_cleanup(config->delay_context);
src/server/receiver.h
+14 -2
View File
@@ -4,6 +4,7 @@
#include "config.h"
#include "file.h"
#include "file_receive.h"
#include "protocol.h"
#include <stdbool.h>
#include <time.h>
@@ -21,6 +22,12 @@ typedef struct {
typedef bool (*ReceiverSuccessFrame)(int fd, void* context);
/* Records that a --max-delete commit stopped with extras left over, so the
caller's terminal success frame can carry STATUS_DELETE_LIMIT instead of
STATUS_OK. The commit runs on the receiver thread, so the flag is stored in
the sink's own context rather than in a shared global. */
typedef void (*ReceiverNoteDeleteLimit)(void* context);
typedef struct {
ReceiverFileSink store_file;
void* context;
@@ -28,13 +35,18 @@ typedef struct {
bool send_success;
/* Emits the end-of-transfer success frame. When the sender requested
--remove-source-files this includes one per-file status per processed
data file followed by the final STATUS_OK; otherwise just STATUS_OK. */
data file followed by the final status; otherwise just the final status. */
ReceiverSuccessFrame send_success_frame;
/* Optional; may be NULL when the sink has no --max-delete handling. */
ReceiverNoteDeleteLimit note_delete_limit;
} ReceiverSink;
bool receiver_outcomes_append(ReceiverOutcomes* outcomes, unsigned char code);
void receiver_outcomes_destroy(ReceiverOutcomes* outcomes);
bool receiver_send_final_success(int fd, const Config* config, const ReceiverOutcomes* outcomes);
/* Send the terminal success frame. `final_status` is usually STATUS_OK, or
STATUS_DELETE_LIMIT when a --max-delete commit was capped. */
bool receiver_send_final_success(int fd, const Config* config, const ReceiverOutcomes* outcomes,
Status final_status);
int receiver_process(Config* config, int file_descriptor, const ReceiverSink* sink);
/* receiver_process with an escape hatch for the commit-style (late) deletion:
src/server/receiver_pipeline.c
+12 -1
View File
@@ -27,6 +27,7 @@ PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue*
context->queued_bytes = 0;
context->max_queue_bytes = 0;
context->deferred_manifest = NULL;
context->delete_limit_reached = false;
atomic_init(&context->cancelled, false);
int init = 0;
if (mtx_init(&context->mutex, mtx_plain) != thrd_success)
@@ -135,6 +136,15 @@ static bool receiver_enqueue_file(File* file, void* context_pointer) {
return pipeline_context_receiver_enqueue_file(context, file);
}
/* Early delete modes (--delete-before/--delete-during) commit the manifest
inside receiver_process_pending on this thread; record a capped commit so
server.c's terminal frame can report STATUS_DELETE_LIMIT. The plain bool is
safe: receive_thread writes it before the main thread joins the thread. */
static void receiver_pipeline_note_delete_limit(void* context_pointer) {
PipelineContextReceiver* context = (PipelineContextReceiver*)context_pointer;
context->delete_limit_reached = true;
}
static void receiver_thread_fail(PipelineContextReceiver* context) {
mtx_lock(&context->mutex);
atomic_store(&context->cancelled, true);
@@ -152,7 +162,8 @@ int receive_thread(void* pipeline_context) {
const Config* config = context->config;
mtx_unlock(&context->mutex);
ReceiverSink sink = {receiver_enqueue_file, context, false, false, NULL};
ReceiverSink sink = {
receiver_enqueue_file, context, false, false, NULL, receiver_pipeline_note_delete_limit};
if (receiver_process_pending((Config*)config, file_descriptor, &sink,
&context->deferred_manifest) != 0) {
receiver_thread_fail(context);
src/server/receiver_pipeline.h
+4
View File
@@ -41,6 +41,10 @@ typedef struct PipelineContextReceiver {
transfer truly succeeded. NULL in the early delete modes (which delete at
the manifest). */
DeleteManifest* deferred_manifest;
/* Set by server.c when the deferred delete commit hit the --max-delete
budget; the terminal success frame then carries STATUS_DELETE_LIMIT
(rsync exit 25) while the transfer itself still succeeds. */
bool delete_limit_reached;
/* P7 Wave D: directory metadata collected by write_thread from received
directory entries. Only write_thread mutates it (before it joins); the
caller (server.c) applies it after the delete/delay-updates phase. */
src/server/server.c
+8 -2
View File
@@ -949,8 +949,13 @@ void handler(int file_descriptor) {
--delay-updates run; the walker skips the staging directory. A
server-contacting --dry-run deletes nothing (no manifest is sent). */
if (context->deferred_manifest) {
if (!manifest_delete_all(config, context->deferred_manifest)) {
DeleteCommitResult deletion = manifest_delete_all(config, context->deferred_manifest);
if (deletion == DELETE_COMMIT_ERROR) {
transfer_ok = false;
} else if (deletion == DELETE_COMMIT_LIMIT_REACHED) {
/* The transfer still succeeds; the terminal frame reports the capped
deletion so the sender exits 25 like rsync. */
context->delete_limit_reached = true;
}
delete_manifest_free(context->deferred_manifest);
context->deferred_manifest = NULL;
@@ -974,7 +979,8 @@ void handler(int file_descriptor) {
dir_metadata_list_apply(&context->dir_times, config->receive_root_directory, config);
}
if (transfer_ok) {
if (!receiver_send_final_success(file_descriptor, config, &context->outcomes))
Status final_status = context->delete_limit_reached ? STATUS_DELETE_LIMIT : STATUS_OK;
if (!receiver_send_final_success(file_descriptor, config, &context->outcomes, final_status))
transfer_ok = false;
} else {
send_error_detail(file_descriptor, "transfer failed on receiver");