diff --git a/src/client/client_cli.c b/src/client/client_cli.c index 6a4399f..29b0bf9 100644 --- a/src/client/client_cli.c +++ b/src/client/client_cli.c @@ -11,9 +11,6 @@ #include #include -char* server_host = "127.0.0.1"; -int server_port = 8080; - static void print_usage(void) { printf("Usage:\n"); printf(" fastsync [options] \n"); @@ -157,10 +154,10 @@ int main(int argc, char* argv[]) { config->use_chunk_serialization = true; log_message(LOG_LEVEL_INFO, "Enabled Chunk Serialization"); } else if (strcmp(argv[i], "--server-host") == 0 && i + 1 < argc) { - free(server_host); - server_host = str_dup(argv[++i]); + free(config->server_host); + config->server_host = str_dup(argv[++i]); } else if (strcmp(argv[i], "--server-port") == 0 && i + 1 < argc) { - server_port = atoi(argv[++i]); + config->server_port = atoi(argv[++i]); } else if (strcmp(argv[i], "--bwlimit") == 0 && i + 1 < argc) { char* end; errno = 0; diff --git a/src/client/client_send.c b/src/client/client_send.c index 8b5bb4a..c247c9d 100644 --- a/src/client/client_send.c +++ b/src/client/client_send.c @@ -98,7 +98,51 @@ static int send_delta(Client* client, File* file, DeltaSignature* sig, Config* c typedef bool (*file_send_fn)(File*, int, bool, int, bool); -static int send_file_incremental(Client* client, File* file, Config* config, file_send_fn send_fn) { +// Send a single file directly (non-incremental path). +static bool send_file_direct(File* file, int fd, bool use_metadata, int compression_level) { + if (!send_status(fd, STATUS_NEXT)) + return false; + return file_send_single_calls(file, fd, use_metadata, compression_level, true); +} + +// Send a single file directly via sendfile (non-incremental path). +static bool send_file_direct_sendfile(File* file, int fd, bool use_metadata) { + if (!send_status(fd, STATUS_NEXT)) + return false; + return file_send_sendfile(file, fd, use_metadata, 0, true); +} + +// Process one file in a chunk: either via incremental check or direct send. +// Returns 0 on success, 1 if skipped (incremental match), -1 on error. +static int send_single_file(Client* client, File* file, Config* config, + bool use_incremental, bool use_sendfile) { + int compression_level = config->use_compression ? config->compression_level : 0; + + if (!use_incremental) { + if (use_sendfile) { + return send_file_direct_sendfile(file, client->file_descriptor, config->use_metadata) + ? 0 : -1; + } + return send_file_direct(file, client->file_descriptor, config->use_metadata, + compression_level) ? 0 : -1; + } + + // Incremental path: use sendfile for the actual data if enabled and no compression + if (use_sendfile) { + DeltaSignature* sig = NULL; + int rc = incremental_check(client, file, &sig); + if (rc == 1) { delta_signature_destroy(sig); return 1; } + if (rc < 0) { delta_signature_destroy(sig); return -1; } + // rc == 0 or rc == 2 (delta not possible with sendfile) + delta_signature_destroy(sig); + // Fall through: send full file via sendfile (pass 0 for compression_level) + if (!file_send_sendfile(file, client->file_descriptor, config->use_metadata, 0, false)) + return -1; + return 0; + } + + // Incremental path with single_calls (supports compression and delta) + file_send_fn send_fn = (file_send_fn)file_send_single_calls; DeltaSignature* sig = NULL; int rc = incremental_check(client, file, &sig); if (rc < 0) { @@ -112,15 +156,12 @@ static int send_file_incremental(Client* client, File* file, Config* config, fil if (rc == 2 && config->use_delta) { int drc = send_delta(client, file, sig, config); delta_signature_destroy(sig); - if (drc == 0) - return 0; - if (drc < 0) - return -1; + if (drc == 0) return 0; + if (drc < 0) return -1; } else { delta_signature_destroy(sig); } - if (!send_fn(file, client->file_descriptor, config->use_metadata, - config->use_compression ? config->compression_level : 0, false)) + if (!send_fn(file, client->file_descriptor, config->use_metadata, compression_level, false)) return -1; return 0; } @@ -142,40 +183,17 @@ int send_chunk(Client* client, Chunk* chunk, Config* config) { return -1; } data_destroy(data); - } else if (config->use_sendfile && !config->use_compression) { - for (int i = 0; i < chunk->element_count; i++) { - if (config->use_incremental) { - int rc = send_file_incremental(client, chunk->items[i], config, - (file_send_fn)file_send_sendfile); - if (rc == 1) - continue; - if (rc < 0) - return -1; - } else { - if (!send_status(client->file_descriptor, STATUS_NEXT)) - return -1; - if (!file_send_sendfile(chunk->items[i], client->file_descriptor, config->use_metadata, 0, - true)) - return -1; - } - } - } else { - for (int i = 0; i < chunk->element_count; i++) { - if (config->use_incremental) { - int rc = send_file_incremental(client, chunk->items[i], config, - (file_send_fn)file_send_single_calls); - if (rc == 1) - continue; - if (rc < 0) - return -1; - } else { - if (!send_status(client->file_descriptor, STATUS_NEXT)) - return -1; - if (!file_send_single_calls(chunk->items[i], client->file_descriptor, config->use_metadata, - config->use_compression ? config->compression_level : 0, true)) - return -1; - } - } + return 0; + } + + bool use_sendfile = config->use_sendfile && !config->use_compression; + for (int i = 0; i < chunk->element_count; i++) { + int rc = send_single_file(client, chunk->items[i], config, + config->use_incremental, use_sendfile); + if (rc == 1) + continue; + if (rc < 0) + return -1; } return 0; } @@ -191,8 +209,10 @@ static int send_chunks_multithreaded(void* pipeline_context) { client = client_connect_ssh(context->config->ssh_destination, context->config->ssh_port); } else if (context->config->use_tls) { client = client_create(); - if (!client || !client_connect_tls(client, server_host, server_port, context->config->tls_cert, - context->config->tls_key, context->config->tls_ca)) { + if (!client || !client_connect_tls(client, context->config->server_host, + context->config->server_port, + context->config->tls_cert, context->config->tls_key, + context->config->tls_ca)) { if (client) client_delete(client); fprintf(stderr, "Error: could not connect to server via TLS\n"); @@ -200,7 +220,8 @@ static int send_chunks_multithreaded(void* pipeline_context) { } } else { client = client_create(); - if (!client || !client_connect(client, server_host, server_port)) { + if (!client || !client_connect(client, context->config->server_host, + context->config->server_port)) { if (client) client_delete(client); fprintf(stderr, "Error: could not connect to server\n"); @@ -338,8 +359,8 @@ int send_files(Config* config) { return 1; } else if (config->use_tls) { client = client_create(); - if (!client || !client_connect_tls(client, server_host, server_port, config->tls_cert, - config->tls_key, config->tls_ca)) { + if (!client || !client_connect_tls(client, config->server_host, config->server_port, + config->tls_cert, config->tls_key, config->tls_ca)) { if (client) client_delete(client); fprintf(stderr, "Error: could not connect to server via TLS\n"); @@ -347,7 +368,7 @@ int send_files(Config* config) { } } else { client = client_create(); - if (!client || !client_connect(client, server_host, server_port)) { + if (!client || !client_connect(client, config->server_host, config->server_port)) { if (client) client_delete(client); fprintf(stderr, "Error: could not connect to server\n"); diff --git a/src/client/client_send.h b/src/client/client_send.h index 83e9e00..e3c484f 100644 --- a/src/client/client_send.h +++ b/src/client/client_send.h @@ -5,9 +5,6 @@ #include "config.h" #include "transport_tcp.h" -extern char* server_host; -extern int server_port; - int send_chunk(Client* client, Chunk* chunk, Config* config); int send_files(Config* config); int send_files_multithreaded(Config* config); diff --git a/src/client/scanner.c b/src/client/scanner.c index c2b61aa..ad56fb2 100644 --- a/src/client/scanner.c +++ b/src/client/scanner.c @@ -55,6 +55,7 @@ static Chunk* chunk_data_to_chunk(ArrayList* chunk_data) { return chunk; } +// Returns: 1 on success, 0 if no more directories in queue, -1 on opendir failure static int open_next_directory(DirectoryScanner* scanner) { if (scanner->current_dir) { closedir(scanner->current_dir); @@ -71,7 +72,7 @@ static int open_next_directory(DirectoryScanner* scanner) { perror("Could not open directory"); free(scanner->current_path); scanner->current_path = NULL; - return 0; + return -1; } return 1; } @@ -82,8 +83,11 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) { while (1) { if (scanner->current_dir == NULL) { - if (!open_next_directory(scanner)) + int ret = open_next_directory(scanner); + if (ret == 0) break; + if (ret < 0) + continue; } struct dirent* entry = readdir(scanner->current_dir); diff --git a/src/shared/compression.c b/src/shared/compression.c index aa27ecb..3641859 100644 --- a/src/shared/compression.c +++ b/src/shared/compression.c @@ -7,7 +7,6 @@ #define INITIAL_DECOMPRESS_BUF_SIZE (1024 * 1024) Data* data_compress(Data* data_to_compress, int compression_level) { - (void)compression_level; log_message(LOG_LEVEL_DEBUG, "Starting to compress data"); size_t dst_size = ZSTD_compressBound(data_to_compress->size); Data* compressed_data = data_create_empty(dst_size); @@ -21,6 +20,14 @@ Data* data_compress(Data* data_to_compress, int compression_level) { return NULL; } + size_t zret = ZSTD_CCtx_setParameter(cctx, ZSTD_c_compressionLevel, compression_level); + if (ZSTD_isError(zret)) { + log_message(LOG_LEVEL_ERROR, "Failed to set compression level: %s", ZSTD_getErrorName(zret)); + ZSTD_freeCCtx(cctx); + data_destroy(compressed_data); + return NULL; + } + ZSTD_inBuffer input = {data_to_compress->data, data_to_compress->size, 0}; ZSTD_outBuffer output = {compressed_data->data, dst_size, 0}; diff --git a/src/shared/config.c b/src/shared/config.c index d16761d..cbfe502 100644 --- a/src/shared/config.c +++ b/src/shared/config.c @@ -45,6 +45,8 @@ Config* config_create(char* version, char* send_directory, char* receive_directo config->tls_cert = NULL; config->tls_key = NULL; config->tls_ca = NULL; + config->server_host = str_dup("127.0.0.1"); + config->server_port = 8080; return config; } @@ -88,6 +90,7 @@ void config_delete(Config* config) { free(config->tls_cert); free(config->tls_key); free(config->tls_ca); + free(config->server_host); free(config); } @@ -110,7 +113,7 @@ bool config_send(int file_descriptor, const Config* config) { return false; if (!send_int(file_descriptor, config->compression_level)) return false; - if (!send_int(file_descriptor, (int)config->chunk_size)) + if (!send_n_data(file_descriptor, &config->chunk_size, sizeof(config->chunk_size))) return false; if (!send_int(file_descriptor, config->use_sendfile)) return false; @@ -183,9 +186,8 @@ Config* config_receive(int file_descriptor) { if (!receive_int(file_descriptor, &tmp)) goto error; config->compression_level = tmp; - if (!receive_int(file_descriptor, &tmp)) + if (!receive_n_data(file_descriptor, &config->chunk_size, sizeof(config->chunk_size))) goto error; - config->chunk_size = (unsigned long long)tmp; if (!receive_int(file_descriptor, &tmp)) goto error; config->use_sendfile = tmp; @@ -218,6 +220,8 @@ Config* config_receive(int file_descriptor) { config->tls_cert = NULL; config->tls_key = NULL; config->tls_ca = NULL; + config->server_host = str_dup("127.0.0.1"); + config->server_port = 8080; if (!send_status(file_descriptor, STATUS_OK)) goto error; return config; @@ -226,6 +230,7 @@ error: free(config->version); free(config->send_directory); free(config->receive_root_directory); + free(config->server_host); free(config); return NULL; } diff --git a/src/shared/config.h b/src/shared/config.h index c1f025b..1ed85fa 100644 --- a/src/shared/config.h +++ b/src/shared/config.h @@ -35,6 +35,8 @@ typedef struct Config { uint32_t delta_block_size; unsigned long long delta_max_file_size; bool use_tls; + char* server_host; + int server_port; char* tls_cert; char* tls_key; char* tls_ca; diff --git a/src/shared/file.c b/src/shared/file.c index 0fb81ff..6d14238 100644 --- a/src/shared/file.c +++ b/src/shared/file.c @@ -409,33 +409,48 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) { } bool to_disk(const char* path, const void* data, unsigned long long data_size) { - char* directory = str_dup(path); - char* dir_to_free = directory; - directory = dirname(directory); - if (!mkdir_r(directory)) { - free(dir_to_free); + // dirname() may modify its argument and may return a pointer to static storage. + // We must use a copy of the result to be safe. + char* path_dup = str_dup(path); + if (!path_dup) return false; + char* dir_result = dirname(path_dup); + char* directory = str_dup(dir_result); + free(path_dup); + if (!directory) + return false; + + bool ok = true; + if (!mkdir_r(directory)) { + ok = false; + goto done; } FILE* file_pointer = fopen(path, "wb"); if (file_pointer == NULL) { perror("Could not open File"); - free(dir_to_free); - return false; + ok = false; + goto done; } if (fwrite(data, 1, data_size, file_pointer) != data_size) { perror("Failed to write all data to disk"); fclose(file_pointer); - free(dir_to_free); - return false; + ok = false; + goto done; } fclose(file_pointer); - free(dir_to_free); - return true; + +done: + free(directory); + return ok; } bool file_send_sendfile(File* file, int file_descriptor, bool use_metadata, int compression_level, bool send_path) { - (void)compression_level; + // sendfile is incompatible with compression (kernel zero-copy). + // If compression is requested, fall back to the regular send path. + if (compression_level > 0) + return file_send_single_calls(file, file_descriptor, use_metadata, compression_level, send_path); + if (send_path && !send_str(file_descriptor, file->path)) return false; if (use_metadata && !metadata_send(file_descriptor, file->metadata)) diff --git a/src/shared/metadata.c b/src/shared/metadata.c index b1dbb57..efee84a 100644 --- a/src/shared/metadata.c +++ b/src/shared/metadata.c @@ -1,6 +1,8 @@ #include "metadata.h" #include "file.h" +#include "log.h" #include "protocol.h" +#include #include #include #include @@ -96,9 +98,10 @@ FileMetadata* metadata_receive(int file_descriptor, int* ok) { void file_restore_metadata(const char* path, FileMetadata* metadata) { if (metadata == NULL) return; - chmod(path, metadata->mode & 07777); - int chown_ret = chown(path, metadata->uid, metadata->gid); - (void)chown_ret; + if (chmod(path, metadata->mode & 07777) != 0) + log_message(LOG_LEVEL_WARNING, "Failed to chmod %s: %s", path, strerror(errno)); + if (chown(path, metadata->uid, metadata->gid) != 0) + log_message(LOG_LEVEL_WARNING, "Failed to chown %s: %s", path, strerror(errno)); struct timespec times[2]; times[0].tv_sec = 0; times[0].tv_nsec = UTIME_OMIT; diff --git a/src/shared/transport_ssh.c b/src/shared/transport_ssh.c index 9eb5264..71f1e3d 100644 --- a/src/shared/transport_ssh.c +++ b/src/shared/transport_ssh.c @@ -1,4 +1,5 @@ #include "transport_ssh.h" +#include "utils.h" #include #include #include @@ -8,39 +9,58 @@ #include typedef struct { - char user[256]; - char host[256]; - char remote_path[4096]; + char* user; + char* host; + char* remote_path; } RemoteDest; +static void remote_dest_destroy(RemoteDest* r) { + free(r->user); + free(r->host); + free(r->remote_path); +} + static int parse_remote_dest(const char* dest, RemoteDest* r) { + memset(r, 0, sizeof(*r)); const char* colon = strchr(dest, ':'); if (!colon) return -1; - size_t remote_path_len = strlen(colon + 1); - if (remote_path_len >= sizeof(r->remote_path)) + r->remote_path = str_dup(colon + 1); + if (!r->remote_path) return -1; - memcpy(r->remote_path, colon + 1, remote_path_len + 1); const char* at = memchr(dest, '@', colon - dest); if (at) { size_t user_len = at - dest; - if (user_len >= sizeof(r->user)) + r->user = malloc(user_len + 1); + if (!r->user) { + remote_dest_destroy(r); return -1; + } memcpy(r->user, dest, user_len); r->user[user_len] = '\0'; size_t host_len = colon - at - 1; - if (host_len >= sizeof(r->host)) + r->host = malloc(host_len + 1); + if (!r->host) { + remote_dest_destroy(r); return -1; + } memcpy(r->host, at + 1, host_len); r->host[host_len] = '\0'; } else { - r->user[0] = '\0'; - size_t host_len = colon - dest; - if (host_len >= sizeof(r->host)) + r->user = str_dup(""); + if (!r->user) { + remote_dest_destroy(r); return -1; + } + size_t host_len = colon - dest; + r->host = malloc(host_len + 1); + if (!r->host) { + remote_dest_destroy(r); + return -1; + } memcpy(r->host, dest, host_len); r->host[host_len] = '\0'; } @@ -57,6 +77,7 @@ Client* client_connect_ssh(const char* destination, int port) { int sv[2]; if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0) { perror("socketpair failed"); + remote_dest_destroy(&r); return NULL; } @@ -71,6 +92,7 @@ Client* client_connect_ssh(const char* destination, int port) { perror("pipe failed"); close(sv[0]); close(sv[1]); + remote_dest_destroy(&r); return NULL; } @@ -81,6 +103,7 @@ Client* client_connect_ssh(const char* destination, int port) { close(sv[1]); close(exec_pipe[0]); close(exec_pipe[1]); + remote_dest_destroy(&r); return NULL; } @@ -88,6 +111,8 @@ Client* client_connect_ssh(const char* destination, int port) { close(sv[0]); close(exec_pipe[0]); fcntl(exec_pipe[1], F_SETFD, FD_CLOEXEC); + // Child doesn't need the RemoteDest strings + remote_dest_destroy(&r); if (sv[1] != STDIN_FILENO) dup2(sv[1], STDIN_FILENO); @@ -97,7 +122,7 @@ Client* client_connect_ssh(const char* destination, int port) { close(sv[1]); char ssh_user[512]; - if (r.user[0] != '\0') + if (r.user && r.user[0] != '\0') snprintf(ssh_user, sizeof(ssh_user), "%s@%s", r.user, r.host); else snprintf(ssh_user, sizeof(ssh_user), "%s", r.host); @@ -138,10 +163,13 @@ Client* client_connect_ssh(const char* destination, int port) { if (n > 0) { close(sv[0]); waitpid(pid, NULL, 0); + remote_dest_destroy(&r); fprintf(stderr, "Error: could not launch 'fastsync-server --stdio' on remote\n"); return NULL; } + remote_dest_destroy(&r); + Client* client = malloc(sizeof(Client)); if (client == NULL) { close(sv[0]); diff --git a/src/shared/utils.c b/src/shared/utils.c index 0846e72..aa7d313 100644 --- a/src/shared/utils.c +++ b/src/shared/utils.c @@ -26,7 +26,8 @@ bool mkdir_r(const char* path) { path_current[0] = '\0'; } const char* delimiter = "/"; - const char* part = strtok(path_duplicate, delimiter); + char* saveptr; + const char* part = strtok_r(path_duplicate, delimiter, &saveptr); bool ok = true; while (part != NULL) { strcpy(path_current_position, part); @@ -41,7 +42,7 @@ bool mkdir_r(const char* path) { break; } } - part = strtok(NULL, delimiter); + part = strtok_r(NULL, delimiter, &saveptr); } free(path_duplicate); free(path_current); @@ -81,10 +82,22 @@ bool glob_match(const char* pattern, const char* str) { return *str == '\0'; } +static bool is_dir_in_manifest(const char* rel_path, ArrayList* manifest) { + size_t len = strlen(rel_path); + for (int i = 0; i < manifest->size; i++) { + const char* entry = (const char*)manifest->items[i]; + // Check if entry starts with rel_path + '/' + if (strncmp(entry, rel_path, len) == 0 && entry[len] == '/') + return true; + } + return false; +} + static void delete_extras_walk(const char* abs_path, const char* rel_path, ArrayList* manifest) { DIR* dir = opendir(abs_path); if (!dir) return; + bool all_removed = true; struct dirent* entry; while ((entry = readdir(dir)) != NULL) { if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) @@ -99,6 +112,10 @@ static void delete_extras_walk(const char* abs_path, const char* rel_path, Array } if (S_ISDIR(st.st_mode)) { delete_extras_walk(child_abs, child_rel, manifest); + // After recursion, try to remove the subdirectory if it's now empty + if (rmdir(child_abs) != 0) { + all_removed = false; + } } else { // Check if relative path is in manifest bool found = false; @@ -111,13 +128,19 @@ static void delete_extras_walk(const char* abs_path, const char* rel_path, Array if (!found) { unlink(child_abs); fprintf(stderr, " Deleted: %s\n", child_rel); + } else { + all_removed = false; } } free(child_abs); free(child_rel); } closedir(dir); - rmdir(abs_path); + // Only remove the directory itself if it is not in the manifest + // and contained no kept entries. + if (all_removed && rel_path[0] != '\0' && !is_dir_in_manifest(rel_path, manifest)) { + rmdir(abs_path); + } } void delete_extras(const char* dest_root, ArrayList* manifest) {