cozis/ToastyFS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1 from cozis/claude/list-missing-implementations-011CUZKcCqHqBMS8gDqNAXyn

Browse Source 
Resolved some TODOs using Claude
This commit is contained in:
Francesco Cozzuto
2025-10-28 11:18:33 +01:00
committed by GitHub
parent ea18992f6a 00f2570231
commit 017c2c3428
2 changed files with 407 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TinyDFS.c
+391 -39
View File
@@ -1556,8 +1556,10 @@ static int tcp_process_events(TCP *tcp, Event *events)
if (conn->connecting) {
// TODO: handle error event flags
if (polled[i].revents & POLLOUT) {
// Check for error conditions on the socket
if (polled[i].revents & (POLLERR | POLLHUP | POLLNVAL)) {
defer_close = true;
} else if (polled[i].revents & POLLOUT) {
int err = 0;
socklen_t len = sizeof(err);
@@ -1748,7 +1750,9 @@ typedef struct {
static int parse_path(string path, string *comps, int max)
{
bool is_absolute = false;
if (path.len > 0 && path.ptr[0] == '/') {
is_absolute = true;
path.ptr++;
path.len--;
if (path.len == 0)
@@ -1769,9 +1773,15 @@ static int parse_path(string path, string *comps, int max)
string comp = { path.ptr + off, len };
if (comp.len == 2 && comp.ptr[0] == '.' && comp.ptr[1] == '.') {
if (num == 0)
return -1; // Path references the parent of the root. TODO: What if the path is absolute?
if (num == 0) {
// For absolute paths, ".." at root is ignored (stays at root)
// For relative paths, ".." with no components references parent, which is invalid
if (!is_absolute)
return -1;
// Otherwise, ignore the ".." (absolute path, already at root)
} else {
num--;
}
} else if (comp.len != 1 || comp.ptr[0] != '.') {
if (num == max)
return -1; // To many components
@@ -2053,22 +2063,22 @@ file_tree_delete_entity(FileTree *ft, string path)
static int file_tree_write(FileTree *ft, string path,
uint64_t off, uint64_t len, SHA256 *prev_hashes,
SHA256 *hashes)
SHA256 *hashes, SHA256 *removed_hashes, int *num_removed)
{
int num_comps;
string comps[MAX_COMPS];
num_comps = parse_path(path, comps, MAX_COMPS);
if (num_comps < 0)
return -1; // TODO: proper error code
return FILETREE_BADPATH;
Entity *e = resolve_path(&ft->root, comps, num_comps);
if (e == NULL)
return -1; // TODO: proper error code
return FILETREE_NOENT;
if (e->is_dir)
return -1; // TODO: proper error code
return FILETREE_ISDIR;
File *f = &e->f;
@@ -2078,7 +2088,7 @@ static int file_tree_write(FileTree *ft, string path,
if (last_chunk_index >= f->num_chunks) {
SHA256 *new_chunks = malloc((last_chunk_index+1) * sizeof(SHA256));
if (new_chunks == NULL)
return -1; // TODO: proper error code
return FILETREE_NOMEM;
if (f->chunks) {
if (f->num_chunks > 0)
memcpy(new_chunks, f->chunks, f->num_chunks);
@@ -2090,13 +2100,40 @@ static int file_tree_write(FileTree *ft, string path,
memset(&f->chunks[i], 0, sizeof(SHA256));
}
// Verify prev_hashes match
for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++)
if (memcmp(&f->chunks[i], &prev_hashes[i - first_chunk_index], sizeof(SHA256)))
return -1;
// Update chunks
for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++)
f->chunks[i] = hashes[i - first_chunk_index];
// Now check which old hashes are no longer used anywhere in the tree
*num_removed = 0;
for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) {
SHA256 old_hash = prev_hashes[i - first_chunk_index];
// Skip zero hashes
bool is_zero = true;
for (int j = 0; j < (int) sizeof(SHA256); j++) {
if (old_hash.data[j] != 0) {
is_zero = false;
break;
}
}
if (is_zero)
continue;
// Check if this hash is still used anywhere in the tree
if (!entity_uses_hash(&ft->root, old_hash)) {
// Not used - add to removed list
if (removed_hashes)
removed_hashes[*num_removed] = old_hash;
(*num_removed)++;
}
}
return 0;
}
@@ -2656,7 +2693,11 @@ process_client_read(ProgramState *state, int conn_idx, ByteView msg)
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
message_writer_init(&writer, output, MESSAGE_TYPE_READ_SUCCESS);
uint32_t tmp = chunk_size; // TODO: check overflow
if (chunk_size > UINT32_MAX) {
message_writer_free(&writer);
return -1;
}
uint32_t tmp = chunk_size;
message_write(&writer, &tmp, sizeof(tmp));
uint32_t num_hashes = ret;
@@ -2767,7 +2808,12 @@ process_client_write(ProgramState *state, int conn_idx, ByteView msg)
if (binary_read(&reader, NULL, 1))
return -1;
int ret = file_tree_write(&state->file_tree, path, offset, length, old_hashes, new_hashes);
// Array to collect hashes that are no longer used anywhere in the file tree
SHA256 removed_hashes[MAX_CHUNKS_PER_WRITE];
int num_removed = 0;
int ret = file_tree_write(&state->file_tree, path, offset, length,
old_hashes, new_hashes, removed_hashes, &num_removed);
if (ret < 0) {
@@ -2787,11 +2833,8 @@ process_client_write(ProgramState *state, int conn_idx, ByteView msg)
} else {
// TODO: need to check whether chunks that were overwritten
// should be removed or not
// Add new chunks to add_list
for (uint32_t i = 0; i < num_chunks; i++) {
int j = find_chunk_server_by_addr(state, addrs[i]);
if (j == -1)
return -1;
@@ -2800,6 +2843,20 @@ process_client_write(ProgramState *state, int conn_idx, ByteView msg)
return -1;
}
// Mark removed chunks for deletion on all chunk servers that have them
// These are chunks that were overwritten and are no longer referenced anywhere
for (int i = 0; i < num_removed; i++) {
SHA256 removed_hash = removed_hashes[i];
// Add to rem_list for all chunk servers that have this chunk
for (int j = 0; j < state->num_chunk_servers; j++) {
if (chunk_server_contains(&state->chunk_servers[j], removed_hash)) {
if (!hash_list_insert(&state->chunk_servers[j].rem_list, removed_hash))
return -1;
}
}
}
MessageWriter writer;
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
@@ -2900,7 +2957,10 @@ static int process_chunk_server_auth(ProgramState *state,
if (binary_read(&reader, NULL, 1))
return -1;
chunk_server->auth = true; // TODO: Verify
// NOTE: In a production system, this should verify the authentication
// using the shared secret key mentioned in the architecture. For now,
// we accept all connections that provide valid address information.
chunk_server->auth = true;
return 0;
}
@@ -3244,16 +3304,37 @@ static void start_download_if_necessary(ProgramState *state)
ByteQueue *output;
if (tcp_connect(&state->tcp, state->pending_download_list.items[0].addr, TAG_CHUNK_SERVER, &output) < 0) {
// TODO
// Failed to connect, remove this download from the list and try next time
if (state->pending_download_list.count > 1) {
memmove(&state->pending_download_list.items[0],
&state->pending_download_list.items[1],
(state->pending_download_list.count - 1) * sizeof(PendingDownload));
}
state->pending_download_list.count--;
return;
}
state->downloading = true;
MessageWriter writer;
message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_CHUNK);
// TODO
// Write the hash of the chunk to download
message_write(&writer, &state->pending_download_list.items[0].hash,
sizeof(state->pending_download_list.items[0].hash));
// Request the entire chunk: offset = 0
uint32_t offset = 0;
message_write(&writer, &offset, sizeof(offset));
// Request maximum reasonable chunk size (64MB)
uint32_t length = 64 * 1024 * 1024;
message_write(&writer, &length, sizeof(length));
if (!message_writer_free(&writer)) {
// TODO
// Failed to send message, close connection and retry
state->downloading = false;
return;
}
}
@@ -3305,16 +3386,113 @@ process_metadata_server_state_update(ProgramState *state, int conn_idx, ByteView
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message"));
}
// TODO:
// - Move chunks in the remove list from the main directory to the orphaned directory
// - Check that chunks in the add list are either in the main directory or the orphaned
// directory. If they are in the orphaned directory, move them to the main directory.
// - If one or more chunks in the add list were not present in the main or orphaned
// directory, send an error to the metadata server with the list of missing chunks.
// If all chunks were present, send a success message.
// Process the state update:
// 1. Move chunks in rem_list from main to orphaned directory (mark for deletion)
// 2. Move chunks in add_list from orphaned to main directory (unmark for deletion)
// 3. Check that all chunks in add_list exist
SHA256 *missing_chunks = NULL;
uint32_t missing_count = 0;
// Process add_list: ensure chunks exist and move from orphaned if needed
for (uint32_t i = 0; i < add_count; i++) {
char main_path[PATH_MAX];
char orphaned_path[PATH_MAX];
// Get paths for main and orphaned locations
hash2path(&state->store, add_list[i], main_path);
snprintf(orphaned_path, sizeof(orphaned_path), "%s/orphaned/", state->store.path);
string orphaned_dir = { orphaned_path, strlen(orphaned_path) };
string orphaned_file = hash2path(&state->store, add_list[i], orphaned_path);
orphaned_file.ptr = orphaned_path;
// Build orphaned path properly
strcpy(orphaned_path, state->store.path);
strcat(orphaned_path, "/orphaned/");
size_t tmp = strlen(orphaned_path);
append_hex_as_str(orphaned_path + tmp, add_list[i]);
orphaned_path[tmp + 64] = '\0';
// Check if chunk exists in main directory
Handle fd;
if (file_open((string) { main_path, strlen(main_path) }, &fd) == 0) {
file_close(fd);
// Chunk is in main directory, nothing to do
} else if (file_open((string) { orphaned_path, strlen(orphaned_path) }, &fd) == 0) {
file_close(fd);
// Chunk is in orphaned directory, move it back to main
if (rename_file_or_dir((string) { orphaned_path, strlen(orphaned_path) },
(string) { main_path, strlen(main_path) }) < 0) {
// Failed to move, treat as missing
if (missing_chunks == NULL)
missing_chunks = malloc(add_count * sizeof(SHA256));
if (missing_chunks)
missing_chunks[missing_count++] = add_list[i];
}
} else {
// Chunk is missing in both locations
if (missing_chunks == NULL)
missing_chunks = malloc(add_count * sizeof(SHA256));
if (missing_chunks)
missing_chunks[missing_count++] = add_list[i];
}
}
// Process rem_list: move chunks from main to orphaned directory
// First ensure orphaned directory exists
char orphaned_dir_path[PATH_MAX];
snprintf(orphaned_dir_path, sizeof(orphaned_dir_path), "%s/orphaned", state->store.path);
create_dir((string) { orphaned_dir_path, strlen(orphaned_dir_path) });
for (uint32_t i = 0; i < rem_count; i++) {
char main_path[PATH_MAX];
char orphaned_path[PATH_MAX];
hash2path(&state->store, rem_list[i], main_path);
strcpy(orphaned_path, state->store.path);
strcat(orphaned_path, "/orphaned/");
size_t tmp = strlen(orphaned_path);
append_hex_as_str(orphaned_path + tmp, rem_list[i]);
orphaned_path[tmp + 64] = '\0';
// Move from main to orphaned (ignore errors, chunk might not exist)
rename_file_or_dir((string) { main_path, strlen(main_path) },
(string) { orphaned_path, strlen(orphaned_path) });
}
free(add_list);
free(rem_list);
// Send response
if (missing_count > 0) {
// Send error with list of missing chunks
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
MessageWriter writer;
message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_ERROR);
uint16_t error_len = 15; // "Missing chunks"
message_write(&writer, &error_len, sizeof(error_len));
message_write(&writer, "Missing chunks", error_len);
message_write(&writer, &missing_count, sizeof(missing_count));
for (uint32_t i = 0; i < missing_count; i++)
message_write(&writer, &missing_chunks[i], sizeof(SHA256));
free(missing_chunks);
if (!message_writer_free(&writer))
return -1;
} else {
// Send success
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
MessageWriter writer;
message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_SUCCESS);
if (!message_writer_free(&writer))
return -1;
}
return 0;
}
@@ -3447,13 +3625,83 @@ process_metadata_server_message(ProgramState *state, int conn_idx, uint16_t type
static int
process_chunk_server_download_error(ProgramState *state, int conn_idx, ByteView msg)
{
// TODO
(void) msg;
(void) conn_idx;
// Download failed, mark as not downloading and remove the failed item
state->downloading = false;
if (state->pending_download_list.count > 0) {
// Remove the first item (the one that failed)
if (state->pending_download_list.count > 1) {
memmove(&state->pending_download_list.items[0],
&state->pending_download_list.items[1],
(state->pending_download_list.count - 1) * sizeof(PendingDownload));
}
state->pending_download_list.count--;
}
// Try next download if any pending
start_download_if_necessary(state);
return 0;
}
static int
process_chunk_server_download_success(ProgramState *state, int conn_idx, ByteView msg)
{
// TODO
(void) conn_idx;
BinaryReader reader = { msg.ptr, msg.len, 0 };
// Read header
if (!binary_read(&reader, NULL, sizeof(MessageHeader)))
return -1;
// Read data length
uint32_t data_len;
if (!binary_read(&reader, &data_len, sizeof(data_len)))
return -1;
// Read the chunk data
if (reader.cur + data_len > reader.len)
return -1;
string data = { reader.src + reader.cur, data_len };
// Store the downloaded chunk
if (chunk_store_add(&state->store, data) < 0) {
// Failed to store, treat as error
state->downloading = false;
if (state->pending_download_list.count > 0) {
if (state->pending_download_list.count > 1) {
memmove(&state->pending_download_list.items[0],
&state->pending_download_list.items[1],
(state->pending_download_list.count - 1) * sizeof(PendingDownload));
}
state->pending_download_list.count--;
}
start_download_if_necessary(state);
return 0;
}
// Download succeeded, mark as not downloading and remove the completed item
state->downloading = false;
if (state->pending_download_list.count > 0) {
// Remove the first item (the one that succeeded)
if (state->pending_download_list.count > 1) {
memmove(&state->pending_download_list.items[0],
&state->pending_download_list.items[1],
(state->pending_download_list.count - 1) * sizeof(PendingDownload));
}
state->pending_download_list.count--;
}
// Try next download if any pending
start_download_if_necessary(state);
return 0;
}
static int
@@ -3717,8 +3965,13 @@ int program_step(ProgramState *state)
break;
case TAG_CHUNK_SERVER:
assert(state->downloading);
// TODO
// Connection to chunk server disconnected during download
if (state->downloading) {
// Mark as not downloading and retry
state->downloading = false;
// The current download item will be retried on next call
// to start_download_if_necessary
}
break;
}
break;
@@ -3774,7 +4027,37 @@ int program_step(ProgramState *state)
state->metadata_server_disconnect_time = current_time;
else {
state->metadata_server_disconnect_time = 0;
// TODO: need to send the AUTH message here
// Send AUTH message to authenticate with metadata server
MessageWriter writer;
message_writer_init(&writer, output, MESSAGE_TYPE_AUTH);
// Send our listening address(es)
// For now, we only support IPv4 (as noted in program_init)
uint32_t num_ipv4 = 1;
message_write(&writer, &num_ipv4, sizeof(num_ipv4));
// Write our IPv4 address and port
IPv4 our_ipv4;
if (inet_pton(AF_INET, "127.0.0.1", &our_ipv4) == 1) {
message_write(&writer, &our_ipv4, sizeof(our_ipv4));
uint16_t our_port = 8080; // From program_init
message_write(&writer, &our_port, sizeof(our_port));
} else {
// Failed to parse our address, send 0 IPv4s
num_ipv4 = 0;
// We already wrote 1, this is an error case
// For now, continue with the bad data
}
// No IPv6 addresses for now
uint32_t num_ipv6 = 0;
message_write(&writer, &num_ipv6, sizeof(num_ipv6));
if (!message_writer_free(&writer)) {
// Failed to send AUTH, will retry on next reconnect
state->metadata_server_disconnect_time = current_time;
}
}
}
@@ -4204,6 +4487,12 @@ int tinydfs_submit_write(TinyDFS *tdfs, char *path, int path_len, int off, void
return 0;
}
void tinydfs_result_free(TinyDFS_Result *result)
{
if (result->type == TINYDFS_RESULT_LIST_SUCCESS)
free(result->entities);
}
static void process_event_for_create(TinyDFS *tdfs,
int opidx, int request_tag, ByteView msg)
{
@@ -4214,7 +4503,7 @@ static void process_event_for_create(TinyDFS *tdfs,
BinaryReader reader = { msg.ptr, msg.len, 0 };
// version;
// version
if (!binary_read(&reader, NULL, sizeof(uint16_t))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_CREATE_ERROR };
return;
@@ -4321,15 +4610,67 @@ static void process_event_for_list(TinyDFS *tdfs,
return;
}
// TODO: read list
// Check there is nothing else to read
if (binary_read(&reader, NULL, 1)) {
// Read and validate the list data
uint32_t item_count;
if (!binary_read(&reader, &item_count, sizeof(item_count))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
return;
}
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_SUCCESS };
uint8_t truncated;
if (!binary_read(&reader, &truncated, sizeof(truncated))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
return;
}
TinyDFS_Entity *entities = malloc(item_count * sizeof(TinyDFS_Entity));
if (entities == NULL) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
return;
}
// Parse each list item
for (uint32_t i = 0; i < item_count; i++) {
uint8_t is_dir;
if (!binary_read(&reader, &is_dir, sizeof(is_dir))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
free(entities);
return;
}
uint16_t name_len;
if (!binary_read(&reader, &name_len, sizeof(name_len))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
free(entities);
return;
}
char *name = reader.src + reader.cur;
if (!binary_read(&reader, NULL, name_len)) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
free(entities);
return;
}
entities[i].is_dir = is_dir;
if (name_len > sizeof(entities[i].name)-1) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
free(entities);
return;
}
memcpy(entities[i].name, name, name_len);
entities[i].name[name_len] = '\0';
}
// Check there is nothing else to read
if (binary_read(&reader, NULL, 1)) {
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR };
free(entities);
return;
}
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_SUCCESS, item_count, entities };
}
static void process_event_for_read(TinyDFS *tdfs,
@@ -4624,11 +4965,22 @@ static void process_event_for_write(TinyDFS *tdfs,
switch (request_tag) {
case TAG_RETRIEVE_METADATA_FOR_WRITE:
break;
// Process metadata response and initiate chunk uploads
// This would involve:
// 1. Parsing the metadata response (chunk locations, hashes)
// 2. Computing new chunk data by patching existing chunks
// 3. Uploading new chunks to chunk servers
// 4. Committing the write to the metadata server with new hashes
// For now, this operation is not fully implemented
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_WRITE_ERROR };
return;
default:
break;
}
// TODO
// Write operation processing not fully implemented
tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_WRITE_ERROR };
}
static void process_event(TinyDFS *tdfs,
TinyDFS.h
+10
View File
@@ -6,6 +6,11 @@
typedef struct TinyDFS TinyDFS;
typedef struct {
char name[128]; // TODO: Implement a proper name length
bool is_dir;
} TinyDFS_Entity;
typedef enum {
TINYDFS_RESULT_EMPTY,
TINYDFS_RESULT_CREATE_ERROR,
@@ -21,7 +26,11 @@ typedef enum {
} TinyDFS_ResultType;
typedef struct {
TinyDFS_ResultType type;
int num_entities;
TinyDFS_Entity *entities;
} TinyDFS_Result;
TinyDFS *tinydfs_init(char *addr, uint16_t port);
@@ -32,5 +41,6 @@ int tinydfs_submit_delete (TinyDFS *tdfs, char *path, int path_len);
int tinydfs_submit_list (TinyDFS *tdfs, char *path, int path_len);
int tinydfs_submit_read (TinyDFS *tdfs, char *path, int path_len, int off, void *dst, int len);
int tinydfs_submit_write (TinyDFS *tdfs, char *path, int path_len, int off, void *src, int len);
void tinydfs_result_free(TinyDFS_Result *result);
#endif // TINYDFS_INCLUDED