commit f3f30b82b058e173347141ecf1ccdebb3a6b2a16 from: Isaac Meerleo date: Fri Jul 17 23:01:10 2026 UTC Persist transactional System epochs commit - 3f83e196092e9ddcb3fcece455be1b57ab878757 commit + f3f30b82b058e173347141ecf1ccdebb3a6b2a16 blob - 998feba74304e5deefac478cc6ac46c6bf9904d1 blob + 5dfc92e6fd593cc61f756f4fca14223e95f8dd8e --- regress/journal/Makefile +++ regress/journal/Makefile @@ -1,5 +1,6 @@ PROG= journal_test -SRCS= journal_test.c journal.c msg.c json.c buf.c log.c xmalloc.c +SRCS= journal_test.c journal.c system_epoch.c compaction.c msg.c json.c \ + buf.c utf8.c log.c xmalloc.c .PATH: ${.CURDIR}/../../src/fugu CFLAGS+= -I${.CURDIR}/../../src/fugu blob - 340cd323bd1007b734c43ef347e0e04d1aee6e21 blob + e361bf736cd133685ed8b034cdd4a8229e31a182 --- regress/journal/journal_test.c +++ regress/journal/journal_test.c @@ -29,10 +29,14 @@ #include #include +#include #include +#include #include +#include #include +#include #include #include #include @@ -43,12 +47,16 @@ #include "xmalloc.h" #include "json.h" #include "msg.h" +#include "compaction.h" +#include "system_epoch.h" #include "journal.h" #include "regress.h" static char dir[] = "/tmp/journal_test.XXXXXXXXXX"; +static int valid_records(const char *, int); + static int count(const struct msglist *l) { @@ -112,7 +120,324 @@ write_turn(struct journal *j, int n, const char *u, co journal_turn_end(j, n, oc); } +static int +buf_eq(const struct buf *b, const void *data, size_t len) +{ + return (b->len == len && + (len == 0 || memcmp(b->data, data, len) == 0)); +} + +static int +file_eq(const struct compaction_files *files, enum compaction_file_kind kind, + size_t at, const void *data, size_t len) +{ + const void *path; + size_t pathlen; + + return (compaction_files_get(files, kind, at, &path, &pathlen) == 0 && + pathlen == len && memcmp(path, data, len) == 0); +} + static void +make_epoch(struct system_epoch *epoch, enum system_epoch_reason reason, + const void *summary, size_t summarylen) +{ + static const char base[] = "pinned base"; + static const char personal[] = "pinned personal"; + static const char project[] = "pinned project"; + struct system_epoch_input input; + + memset(&input, 0, sizeof(input)); + input.reason = reason; + input.base.data = base; + input.base.len = sizeof(base) - 1; + input.personal.data = personal; + input.personal.len = sizeof(personal) - 1; + input.project.data = project; + input.project.len = sizeof(project) - 1; + input.summary.data = summary; + input.summary.len = summarylen; + system_epoch_init(epoch); + CHECK(system_epoch_build(epoch, &input) == 0); +} + +static void +append_text_message(struct journal *j, enum msg_role role, const void *text, + size_t len) +{ + struct msg *m; + + m = msg_new(role); + msg_add_text(m, text, len); + journal_message(j, m); + msg_free(m); +} + +static int +listed_summary(const char *id, int64_t *messages, char *preview, + size_t previewsz) +{ + struct session_ent *sessions; + int i, n, found = 0; + + if ((n = journal_list(dir, &sessions)) < 0) + return (-1); + for (i = 0; i < n; i++) { + if (strcmp(sessions[i].id, id) != 0) + continue; + if (messages != NULL) + *messages = sessions[i].messages; + if (preview != NULL && previewsz > 0) + strlcpy(preview, sessions[i].preview, previewsz); + found = 1; + break; + } + free(sessions); + return (found); +} + +static void +test_system_epoch_roundtrip(void) +{ + static const u_char base[] = { 'b', 'a', 's', 'e', 0xff }; + static const u_char personal[] = { 'p', 'e', 'r', 0xfe }; + static const u_char project[] = { 'p', 'r', 'o', 'j', 0xfd }; + static const u_char summary[] = { 's', 'u', 'm', '\0', 0xfc }; + struct system_epoch_input input; + struct system_epoch epoch; + struct compaction_files *files; + struct journal_state state; + struct journal *j; + char id[SESSION_ID_MAX], path[1024]; + + memset(&input, 0, sizeof(input)); + input.reason = SYSTEM_EPOCH_START; + input.base.data = base; + input.base.len = sizeof(base); + input.personal.data = personal; + input.personal.len = sizeof(personal); + input.project.data = project; + input.project.len = sizeof(project); + input.summary.data = summary; + input.summary.len = sizeof(summary); + system_epoch_init(&epoch); + CHECK(system_epoch_build(&epoch, &input) == 0); + files = compaction_files_new(); + CHECK(compaction_files_add(files, COMPACTION_FILE_READ, + "README.md", 9) == 0); + CHECK(compaction_files_add(files, COMPACTION_FILE_READ, + "AUTHORS", 7) == 0); + CHECK(compaction_files_add(files, COMPACTION_FILE_MODIFIED, + "src/z.c", 7) == 0); + CHECK(compaction_files_add(files, COMPACTION_FILE_MODIFIED, + "src/a.c", 7) == 0); + + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &epoch, files, NULL, 0) == 0); + journal_sync(j); + journal_close(j); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(valid_records(path, 2)); + + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.have_epoch); + CHECK(state.epoch.version == SYSTEM_EPOCH_VERSION); + CHECK(state.epoch.reason == SYSTEM_EPOCH_START); + CHECK(buf_eq(&state.epoch.base, epoch.base.data, epoch.base.len)); + CHECK(buf_eq(&state.epoch.personal, epoch.personal.data, + epoch.personal.len)); + CHECK(buf_eq(&state.epoch.project, epoch.project.data, + epoch.project.len)); + CHECK(buf_eq(&state.epoch.summary, epoch.summary.data, + epoch.summary.len)); + CHECK(buf_eq(&state.epoch.full, epoch.full.data, epoch.full.len)); + CHECK(buf_eq(&state.epoch.restricted, epoch.restricted.data, + epoch.restricted.len)); + CHECK(compaction_files_count(state.files, COMPACTION_FILE_READ) == 2); + CHECK(compaction_files_count(state.files, + COMPACTION_FILE_MODIFIED) == 2); + CHECK(file_eq(state.files, COMPACTION_FILE_READ, 0, "AUTHORS", 7)); + CHECK(file_eq(state.files, COMPACTION_FILE_READ, 1, "README.md", 9)); + CHECK(file_eq(state.files, COMPACTION_FILE_MODIFIED, 0, "src/a.c", 7)); + CHECK(file_eq(state.files, COMPACTION_FILE_MODIFIED, 1, "src/z.c", 7)); + CHECK(count(&state.messages) == 0); + journal_state_free(&state); + compaction_files_free(files); + system_epoch_free(&epoch); +} + +static void +test_epoch_retained_transactions(void) +{ + static const u_char binary_assistant[] = { 'a', '\0', 0xff, 'z' }; + static const u_char binary_result[] = { 'r', '\0', 0xfe, 'z' }; + struct system_epoch start, compact1, compact2; + struct journal_state state; + struct session_ent *sessions; + struct msglist retained1, retained2; + struct journal *j; + struct msg *m; + struct block *b; + char id[SESSION_ID_MAX], listed_preview[80]; + int64_t listed_messages; + int found, i, n; + + make_epoch(&start, SYSTEM_EPOCH_START, NULL, 0); + make_epoch(&compact1, SYSTEM_EPOCH_COMPACT, "summary one", 11); + make_epoch(&compact2, SYSTEM_EPOCH_COMPACT, "summary two", 11); + TAILQ_INIT(&retained1); + m = msg_new(ROLE_USER); + msg_add_text(m, "retained user", 13); + TAILQ_INSERT_TAIL(&retained1, m, entry); + m = msg_new(ROLE_ASSISTANT); + msg_add_text(m, (const char *)binary_assistant, + sizeof(binary_assistant)); + msg_add_tool_use(m, "tool-1", "read", "{\"path\":\"x\"}", 12); + TAILQ_INSERT_TAIL(&retained1, m, entry); + m = msg_new(ROLE_USER); + msg_add_tool_result(m, "tool-1", (const char *)binary_result, + sizeof(binary_result), 1); + TAILQ_INSERT_TAIL(&retained1, m, entry); + + /* An idle prefix replacement publishes its exact retained Projection. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + write_turn(j, 1, "discarded history", "discarded answer", TURN_OK); + CHECK(journal_system_epoch(j, &compact1, NULL, &retained1, 0) == 0); + write_turn(j, 2, "after retained", "after answer", TURN_OK); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.have_epoch && + state.epoch.reason == SYSTEM_EPOCH_COMPACT); + CHECK(count(&state.messages) == 5); + CHECK(strcmp(text_of(&state.messages, 0), "retained user") == 0); + m = TAILQ_FIRST(&state.messages); + m = TAILQ_NEXT(m, entry); + b = TAILQ_FIRST(&m->blocks); + CHECK(b->textlen == sizeof(binary_assistant) && + memcmp(b->text, binary_assistant, sizeof(binary_assistant)) == 0); + b = TAILQ_NEXT(b, entry); + CHECK(b != NULL && b->type == BLOCK_TOOL_USE && + strcmp(b->tool_id, "tool-1") == 0 && + b->tool_input_len == 12 && + memcmp(b->tool_input, "{\"path\":\"x\"}", 12) == 0); + m = TAILQ_NEXT(m, entry); + b = TAILQ_FIRST(&m->blocks); + CHECK(b != NULL && b->type == BLOCK_TOOL_RESULT && b->is_error && + b->result_len == sizeof(binary_result) && + memcmp(b->result, binary_result, sizeof(binary_result)) == 0); + CHECK(strcmp(text_of(&state.messages, 3), "after retained") == 0); + journal_state_free(&state); + + /* Listing folds the same reset/snapshot chronology as restore. */ + found = 0; + n = journal_list(dir, &sessions); + CHECK(n > 0); + for (i = 0; i < n; i++) + if (strcmp(sessions[i].id, id) == 0) { + found = 1; + CHECK(sessions[i].messages == 5); + CHECK(strcmp(sessions[i].preview, "retained user") == 0); + } + CHECK(found); + free(sessions); + + /* Multiple active-Turn replacements fold provisionally. Only the last + * snapshot plus later records publishes when the matching Turn commits. */ + TAILQ_INIT(&retained2); + m = msg_new(ROLE_USER); + msg_add_text(m, "second snapshot", 15); + TAILQ_INSERT_TAIL(&retained2, m, entry); + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + write_turn(j, 1, "old", "old answer", TURN_OK); + journal_turn_begin(j, 2); + append_text_message(j, ROLE_USER, "before first cut", 16); + CHECK(journal_system_epoch(j, &compact1, NULL, &retained1, 2) == 0); + append_text_message(j, ROLE_USER, "between cuts", 12); + CHECK(journal_system_epoch(j, &compact2, NULL, &retained2, 2) == 0); + append_text_message(j, ROLE_ASSISTANT, "after final cut", 15); + journal_turn_end(j, 2, TURN_OK); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.epoch.reason == SYSTEM_EPOCH_COMPACT); + CHECK(buf_eq(&state.epoch.summary, "summary two", 11)); + CHECK(count(&state.messages) == 2); + CHECK(strcmp(text_of(&state.messages, 0), "second snapshot") == 0); + CHECK(strcmp(text_of(&state.messages, 1), "after final cut") == 0); + journal_state_free(&state); + CHECK(listed_summary(id, &listed_messages, listed_preview, + sizeof(listed_preview)) == 1); + CHECK(listed_messages == 2 && + strcmp(listed_preview, "second snapshot") == 0); + + /* An abandoned provisional replacement publishes neither epoch nor + * pending Projection records. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + write_turn(j, 1, "kept", "kept answer", TURN_OK); + journal_turn_begin(j, 2); + append_text_message(j, ROLE_USER, "pending", 7); + CHECK(journal_system_epoch(j, &compact1, NULL, &retained1, 2) == 0); + append_text_message(j, ROLE_ASSISTANT, "also pending", 12); + journal_turn_end(j, 2, TURN_ABANDONED); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.epoch.reason == SYSTEM_EPOCH_START); + CHECK(count(&state.messages) == 2); + CHECK(strcmp(text_of(&state.messages, 0), "kept") == 0); + journal_state_free(&state); + CHECK(listed_summary(id, &listed_messages, listed_preview, + sizeof(listed_preview)) == 1); + CHECK(listed_messages == 2 && strcmp(listed_preview, "kept") == 0); + + /* A cap commits the provisional replacement just like ok. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + journal_turn_begin(j, 7); + CHECK(journal_system_epoch(j, &compact2, NULL, &retained2, 7) == 0); + append_text_message(j, ROLE_ASSISTANT, "cap suffix", 10); + journal_turn_end(j, 7, TURN_CAP); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.epoch.reason == SYSTEM_EPOCH_COMPACT); + CHECK(count(&state.messages) == 2); + journal_state_free(&state); + + /* A crash before turn-end discards the provisional epoch and snapshot. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + write_turn(j, 1, "survives crash", "answer", TURN_OK); + journal_turn_begin(j, 8); + CHECK(journal_system_epoch(j, &compact1, NULL, &retained1, 8) == 0); + append_text_message(j, ROLE_ASSISTANT, "crash suffix", 12); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.epoch.reason == SYSTEM_EPOCH_START); + CHECK(count(&state.messages) == 2); + CHECK(strcmp(text_of(&state.messages, 0), "survives crash") == 0); + journal_state_free(&state); + + msglist_free(&retained1); + msglist_free(&retained2); + system_epoch_free(&start); + system_epoch_free(&compact1); + system_epoch_free(&compact2); +} + +static void test_roundtrip(void) { struct journal *j; @@ -419,7 +744,546 @@ valid_records(const char *path, int want) return (ok && count == want); } +static int +path_load(const char *path, char **datap, size_t *lenp) +{ + struct stat st; + char *data; + ssize_t n; + size_t off = 0; + int fd; + + *datap = NULL; + *lenp = 0; + if (stat(path, &st) == -1 || st.st_size < 0 || + (uintmax_t)st.st_size > SIZE_MAX - 1 || + (fd = open(path, O_RDONLY)) == -1) + return (-1); + data = xmalloc((size_t)st.st_size + 1); + while (off < (size_t)st.st_size) { + n = read(fd, data + off, (size_t)st.st_size - off); + if (n <= 0) { + close(fd); + free(data); + return (-1); + } + off += (size_t)n; + } + close(fd); + data[off] = '\0'; + *datap = data; + *lenp = off; + return (0); +} + +static int +path_store(const char *path, const void *data, size_t len) +{ + const u_char *p = data; + ssize_t n; + size_t off = 0; + int fd; + + if ((fd = open(path, O_WRONLY|O_TRUNC)) == -1) + return (-1); + while (off < len) { + n = write(fd, p + off, len - off); + if (n <= 0) { + close(fd); + return (-1); + } + off += (size_t)n; + } + return (close(fd)); +} + +static char * +last_match(char *data, const char *needle) +{ + char *at, *last = NULL; + + for (at = strstr(data, needle); at != NULL; + at = strstr(at + 1, needle)) + last = at; + return (last); +} + +static int +path_change_after(const char *path, const char *needle, char replacement) +{ + char *at, *data; + size_t len; + int ok; + + if (path_load(path, &data, &len) == -1 || + (at = last_match(data, needle)) == NULL) { + free(data); + return (-1); + } + at[strlen(needle)] = replacement; + ok = path_store(path, data, len); + free(data); + return (ok); +} + +static int +path_replace_same(const char *path, const char *from, const char *to) +{ + char *at, *data = NULL; + size_t len; + int ok; + + if (strlen(from) != strlen(to) || path_load(path, &data, &len) == -1 || + (at = last_match(data, from)) == NULL) { + free(data); + return (-1); + } + memcpy(at, to, strlen(to)); + ok = path_store(path, data, len); + free(data); + return (ok); +} + +static int +path_remove_last(const char *path, const char *needle) +{ + char *at, *data; + size_t len, remove; + int ok; + + if (path_load(path, &data, &len) == -1 || + (at = last_match(data, needle)) == NULL) { + free(data); + return (-1); + } + remove = strlen(needle); + memmove(at, at + remove, len - (size_t)(at - data) - remove); + len -= remove; + ok = path_store(path, data, len); + free(data); + return (ok); +} + +static int +path_insert_last_object(const char *path, const char *addition) +{ + char *at, *data, *next; + size_t addlen, len, off; + int ok; + + if (path_load(path, &data, &len) == -1 || + (at = strrchr(data, '}')) == NULL) { + free(data); + return (-1); + } + off = (size_t)(at - data); + addlen = strlen(addition); + next = xmalloc(len + addlen + 1); + memcpy(next, data, off); + memcpy(next + off, addition, addlen); + memcpy(next + off + addlen, data + off, len - off); + next[len + addlen] = '\0'; + ok = path_store(path, next, len + addlen); + free(next); + free(data); + return (ok); +} + static void +epoch_session(const struct system_epoch *epoch, char *id, char *path) +{ + struct journal *j; + + j = journal_create(dir, id, SESSION_ID_MAX); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, epoch, NULL, NULL, 0) == 0); + journal_close(j); + snprintf(path, 1024, "%s/%s.ndjson", dir, id); +} + +static void +restore_fails(const char *id) +{ + struct journal_state state; + + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == -1); + journal_state_free(&state); +} + +static void +test_epoch_reasons_and_legacy(void) +{ + struct system_epoch start, clear, migration; + struct journal_state state; + struct journal *j; + char id[SESSION_ID_MAX], path[1024], listed_preview[80]; + struct stat st; + off_t partial_size; + int64_t listed_messages; + int fd; + + make_epoch(&start, SYSTEM_EPOCH_START, NULL, 0); + make_epoch(&clear, SYSTEM_EPOCH_CLEAR, NULL, 0); + make_epoch(&migration, SYSTEM_EPOCH_MIGRATION, "legacy summary", 14); + + /* Legacy context and Projection remain readable and migration is + * additive: it installs the exact epoch without resetting messages. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + journal_context(j, "legacy personal", 15, "legacy project", 14); + write_turn(j, 1, "legacy user", "legacy answer", TURN_OK); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(!state.have_epoch); + CHECK(buf_eq(&state.legacy_personal, "legacy personal", 15)); + CHECK(buf_eq(&state.legacy_project, "legacy project", 14)); + CHECK(count(&state.messages) == 2); + journal_state_free(&state); + CHECK(listed_summary(id, &listed_messages, listed_preview, + sizeof(listed_preview)) == 1); + CHECK(listed_messages == 2 && + strcmp(listed_preview, "legacy user") == 0); + j = journal_open(dir, id); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &migration, NULL, NULL, 0) == 0); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.have_epoch && + state.epoch.reason == SYSTEM_EPOCH_MIGRATION); + CHECK(buf_eq(&state.epoch.summary, "legacy summary", 14)); + CHECK(count(&state.messages) == 2); + journal_state_free(&state); + CHECK(listed_summary(id, &listed_messages, listed_preview, + sizeof(listed_preview)) == 1); + CHECK(listed_messages == 2 && + strcmp(listed_preview, "legacy user") == 0); + + /* A complete legacy log plus an incomplete migration tail remains + * legacy, so migration can be retried on the next resume. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + journal_context(j, "retry personal", 14, "retry project", 13); + write_turn(j, 1, "retry user", "retry answer", TURN_OK); + journal_close(j); + j = journal_open(dir, id); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &migration, NULL, NULL, 0) == 0); + journal_close(j); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(stat(path, &st) == 0 && st.st_size > 0); + fd = open(path, O_WRONLY); + CHECK(fd != -1); + CHECK(ftruncate(fd, st.st_size - 1) == 0); + close(fd); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(!state.have_epoch); + CHECK(buf_eq(&state.legacy_personal, "retry personal", 14)); + CHECK(count(&state.messages) == 2); + journal_state_free(&state); + /* Reopening seals, but never truncates, the ignored crash tail. */ + CHECK(stat(path, &st) == 0); + partial_size = st.st_size; + j = journal_open(dir, id); + CHECK(j != NULL); + journal_close(j); + CHECK(stat(path, &st) == 0 && st.st_size > partial_size); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(!state.have_epoch && count(&state.messages) == 2); + journal_state_free(&state); + /* The retried migration is now its own complete replay unit. */ + j = journal_open(dir, id); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &migration, NULL, NULL, 0) == 0); + journal_sync(j); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.have_epoch && + state.epoch.reason == SYSTEM_EPOCH_MIGRATION && + count(&state.messages) == 2); + journal_state_free(&state); + + /* Clear is one atomic epoch+Projection reset. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + write_turn(j, 1, "before clear epoch", "old", TURN_OK); + CHECK(journal_system_epoch(j, &clear, NULL, NULL, 0) == 0); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.have_epoch && state.epoch.reason == SYSTEM_EPOCH_CLEAR); + CHECK(count(&state.messages) == 0); + journal_state_free(&state); + + /* File arrays are optional on input; the retained snapshot is not. */ + epoch_session(&start, id, path); + CHECK(path_remove_last(path, ",\"read_files\":[]") == 0); + CHECK(path_remove_last(path, ",\"modified_files\":[]") == 0); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(state.have_epoch); + CHECK(compaction_files_count(state.files, COMPACTION_FILE_READ) == 0); + CHECK(compaction_files_count(state.files, + COMPACTION_FILE_MODIFIED) == 0); + journal_state_free(&state); + + /* A complete legacy session plus a partial final epoch tail remains + * legacy; the next resume can retry migration/start. */ + epoch_session(&start, id, path); + CHECK(stat(path, &st) == 0 && st.st_size > 0); + fd = open(path, O_WRONLY); + CHECK(fd != -1); + CHECK(ftruncate(fd, st.st_size - 1) == 0); + close(fd); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0); + CHECK(!state.have_epoch && count(&state.messages) == 0); + journal_state_free(&state); + + system_epoch_free(&start); + system_epoch_free(&clear); + system_epoch_free(&migration); +} + +static void +test_epoch_malformed_and_bounds(void) +{ + struct system_epoch start, compact, clear, bad_clear, oversized; + struct system_epoch_input input; + struct compaction_files *files; + struct journal_state state; + struct msglist retained; + struct journal *j; + struct msg *m; + char id[SESSION_ID_MAX], path[1024]; + char *large; + struct stat before, after; + off_t huge = (off_t)JOURNAL_RECORD_MAX + 1; + int fd; + + make_epoch(&start, SYSTEM_EPOCH_START, NULL, 0); + make_epoch(&compact, SYSTEM_EPOCH_COMPACT, "compact", 7); + make_epoch(&clear, SYSTEM_EPOCH_CLEAR, NULL, 0); + make_epoch(&bad_clear, SYSTEM_EPOCH_CLEAR, "not empty", 9); + + /* Stored complete variants must match the rebuilt components exactly. */ + epoch_session(&start, id, path); + CHECK(path_change_after(path, "\"full\":\"", '0') == 0); + restore_fails(id); + + epoch_session(&start, id, path); + CHECK(path_change_after(path, "\"version\":", '9') == 0); + restore_fails(id); + + epoch_session(&start, id, path); + CHECK(path_remove_last(path, ",\"retained\":[]") == 0); + restore_fails(id); + + epoch_session(&start, id, path); + CHECK(path_insert_last_object(path, ",\"version\":1") == 0); + restore_fails(id); + + epoch_session(&start, id, path); + CHECK(path_replace_same(path, "\"reason\":\"start\"", + "\"reason\":\"bogus\"") == 0); + restore_fails(id); + + epoch_session(&start, id, path); + CHECK(path_change_after(path, "\"restricted\":\"", 'g') == 0); + restore_fails(id); + + epoch_session(&start, id, path); + CHECK(path_replace_same(path, "\"base\":\"70", + "\"base\":\"00") == 0); + restore_fails(id); + + /* A stored Clear cannot smuggle forward structured file state. */ + files = compaction_files_new(); + CHECK(files != NULL && compaction_files_add(files, + COMPACTION_FILE_READ, "README.md", 9) == 0); + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, files, NULL, 0) == 0); + journal_close(j); + compaction_files_free(files); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(path_replace_same(path, "\"reason\":\"start\"", + "\"reason\":\"clear\"") == 0); + restore_fails(id); + + /* Complete malformed epoch-looking lines fail, while partial tails were + * covered above. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + journal_close(j); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + fd = open(path, O_WRONLY|O_APPEND); + CHECK(fd != -1); + CHECK(write(fd, "{\"t\":\"system-epoch\"\n", 20) == 20); + close(fd); + restore_fails(id); + CHECK(listed_summary(id, NULL, NULL, 0) == 0); + + /* Epoch/legacy mixing and duplicate or mis-associated boundaries fail. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + journal_context(j, "late", 4, "legacy", 6); + journal_close(j); + restore_fails(id); + + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + journal_context(j, "old", 3, "format", 6); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + journal_close(j); + restore_fails(id); + CHECK(listed_summary(id, NULL, NULL, 0) == 0); + + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + journal_close(j); + restore_fails(id); + + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + journal_turn_begin(j, 2); + CHECK(journal_system_epoch(j, &compact, NULL, NULL, 0) == -1); + CHECK(journal_system_epoch(j, &compact, NULL, NULL, 3) == -1); + journal_turn_end(j, 2, TURN_OK); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0 && state.have_epoch && + state.epoch.reason == SYSTEM_EPOCH_START); + journal_state_free(&state); + + /* Writer-side policy errors do not append a partial record. */ + TAILQ_INIT(&retained); + m = msg_new(ROLE_USER); + msg_add_text(m, "not valid for start", 19); + TAILQ_INSERT_TAIL(&retained, m, entry); + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(stat(path, &before) == 0); + CHECK(journal_system_epoch(j, &start, NULL, &retained, 0) == -1); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 1) == -1); + CHECK(journal_system_epoch(j, &bad_clear, NULL, NULL, 0) == -1); + files = compaction_files_new(); + CHECK(files != NULL && compaction_files_add(files, + COMPACTION_FILE_READ, "README.md", 9) == 0); + CHECK(journal_system_epoch(j, &clear, files, NULL, 0) == -1); + compaction_files_free(files); + CHECK(stat(path, &after) == 0 && after.st_size == before.st_size); + journal_close(j); + msglist_free(&retained); + + /* Retained snapshots must themselves be canonical Projections. */ + TAILQ_INIT(&retained); + m = msg_new(ROLE_USER); + msg_add_tool_result(m, "orphan", "result", 6, 0); + TAILQ_INSERT_TAIL(&retained, m, entry); + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(stat(path, &before) == 0); + CHECK(journal_system_epoch(j, &compact, NULL, &retained, 0) == -1); + CHECK(stat(path, &after) == 0 && after.st_size == before.st_size); + journal_close(j); + msglist_free(&retained); + + /* Replay rejects a complete snapshot whose Tool result no longer + * matches the immediately preceding Assistant call. */ + TAILQ_INIT(&retained); + m = msg_new(ROLE_ASSISTANT); + msg_add_tool_use(m, "pair-1", "read", "{}", 2); + TAILQ_INSERT_TAIL(&retained, m, entry); + m = msg_new(ROLE_USER); + msg_add_tool_result(m, "pair-1", "result", 6, 0); + TAILQ_INSERT_TAIL(&retained, m, entry); + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + CHECK(journal_system_epoch(j, &start, NULL, NULL, 0) == 0); + CHECK(journal_system_epoch(j, &compact, NULL, &retained, 0) == 0); + journal_close(j); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(path_change_after(path, "\"id\":\"", '0') == 0); + restore_fails(id); + msglist_free(&retained); + + /* A valid source epoch whose encoded atomic record exceeds 32 MiB is + * rejected before any bytes are appended. */ + large = xmalloc(6 * 1024 * 1024); + memset(large, 'x', 6 * 1024 * 1024); + memset(&input, 0, sizeof(input)); + input.reason = SYSTEM_EPOCH_START; + input.base.data = large; + input.base.len = 6 * 1024 * 1024; + system_epoch_init(&oversized); + CHECK(system_epoch_build(&oversized, &input) == 0); + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + CHECK(stat(path, &before) == 0); + CHECK(journal_system_epoch(j, &oversized, NULL, NULL, 0) == -1); + CHECK(stat(path, &after) == 0 && after.st_size == before.st_size); + journal_close(j); + system_epoch_free(&oversized); + free(large); + + /* Sparse oversized garbage remains skippable, but an oversized epoch + * record is conspicuous rather than a silent fallback. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + journal_close(j); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + fd = open(path, O_WRONLY|O_TRUNC); + CHECK(fd != -1); + CHECK(write(fd, "{\"t\":\"system-epoch\",", 20) == 20); + CHECK(ftruncate(fd, huge) == 0); + CHECK(lseek(fd, 0, SEEK_END) == huge); + CHECK(write(fd, "\n", 1) == 1); + close(fd); + restore_fails(id); + + /* Oversized ordinary records retain the legacy skip behavior. */ + j = journal_create(dir, id, sizeof(id)); + CHECK(j != NULL); + journal_close(j); + snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id); + fd = open(path, O_WRONLY|O_TRUNC); + CHECK(fd != -1); + CHECK(write(fd, "{\"t\":\"msg\",", 11) == 11); + CHECK(ftruncate(fd, huge) == 0); + CHECK(lseek(fd, 0, SEEK_END) == huge); + CHECK(write(fd, "\n", 1) == 1); + close(fd); + j = journal_open(dir, id); + CHECK(j != NULL); + write_turn(j, 1, "after oversize", "kept", TURN_OK); + journal_close(j); + journal_state_init(&state); + CHECK(journal_restore(dir, id, &state) == 0 && + count(&state.messages) == 2 && + strcmp(text_of(&state.messages, 0), "after oversize") == 0); + journal_state_free(&state); + + system_epoch_free(&start); + system_epoch_free(&compact); + system_epoch_free(&clear); + system_epoch_free(&bad_clear); +} + +static void test_hostile_tool_input(void) { static const char hostile[] = @@ -703,15 +1567,96 @@ test_mismatched_transactions(void) free(list); } +static int +write_full(int fd, const void *data, size_t len) +{ + const u_char *p = data; + ssize_t n; + size_t off = 0; + + while (off < len) { + n = write(fd, p + off, len - off); + if (n == -1 && errno == EINTR) + continue; + if (n <= 0) + return (-1); + off += (size_t)n; + } + return (0); +} + +static int +append_dense_turn(const char *path, size_t pairs) +{ + static const char begin[] = "{\"t\":\"turn-begin\",\"n\":1}\n"; + static const char pair[] = + "{\"t\":\"msg\",\"role\":\"user\",\"blocks\":[{" + "\"bt\":\"text\",\"text\":\"bounded first\"}]}\n" + "{\"t\":\"msg\",\"role\":\"assistant\",\"blocks\":[{" + "\"bt\":\"text\",\"text\":\"bounded answer\"}]}\n"; + static const char end[] = + "{\"t\":\"turn-end\",\"n\":1,\"outcome\":\"ok\"}\n"; + char batch[64 * 1024]; + size_t batch_pairs, i, off; + int fd = -1, ok = -1; + + if ((fd = open(path, O_WRONLY|O_APPEND)) == -1 || + write_full(fd, begin, sizeof(begin) - 1) == -1) + goto done; + batch_pairs = sizeof(batch) / (sizeof(pair) - 1); + while (pairs > 0) { + if (batch_pairs > pairs) + batch_pairs = pairs; + off = 0; + for (i = 0; i < batch_pairs; i++) { + memcpy(batch + off, pair, sizeof(pair) - 1); + off += sizeof(pair) - 1; + } + if (write_full(fd, batch, off) == -1) + goto done; + pairs -= batch_pairs; + } + if (write_full(fd, end, sizeof(end) - 1) == -1) + goto done; + ok = 0; + +done: + if (fd != -1 && close(fd) == -1) + ok = -1; + return (ok); +} + +static int +bounded_listing_ok(const char *bounded_dir, const char *id) +{ + struct session_ent *list; + int found = 0, i, n; + + if ((n = journal_list(bounded_dir, &list)) < 0) + return (0); + for (i = 0; i < n; i++) + if (strcmp(list[i].id, id) == 0 && + list[i].messages == 200000 && + strcmp(list[i].preview, "bounded first") == 0) + found = 1; + free(list); + return (found); +} + static void test_large_append_only_session(void) { struct journal *j; struct msglist conv; struct session_ent *list; - char id[SESSION_ID_MAX], path[1024]; + char bounded_dir[] = + "/tmp/journal_list_bounded.XXXXXXXXXX"; + char id[SESSION_ID_MAX], bounded_id[SESSION_ID_MAX]; + char path[1024], bounded_path[1024]; + struct rlimit limit; + pid_t pid; off_t large = (off_t)65 * 1024 * 1024; - int fd, i, n, found = 0; + int fd, i, n, found = 0, status; /* A sparse oversized garbage record calibrates that replay/listing are * bounded per record rather than rejecting an otherwise recoverable @@ -744,6 +1689,31 @@ test_large_append_only_session(void) } CHECK(found); free(list); + + /* Listing folds arbitrarily many committed messages without retaining + * the live Projection. The old restore-based implementation exhausts + * this child's 32 MiB data limit on the same 200,000-message Journal. */ + CHECK(mkdtemp(bounded_dir) != NULL); + j = journal_create(bounded_dir, bounded_id, sizeof(bounded_id)); + CHECK(j != NULL); + journal_close(j); + CHECK((size_t)snprintf(bounded_path, sizeof(bounded_path), + "%s/%s.ndjson", bounded_dir, bounded_id) < sizeof(bounded_path)); + CHECK(append_dense_turn(bounded_path, 100000) == 0); + pid = fork(); + CHECK(pid != -1); + if (pid == 0) { + limit.rlim_cur = 32 * 1024 * 1024; + limit.rlim_max = 32 * 1024 * 1024; + if (setrlimit(RLIMIT_DATA, &limit) == -1) + _exit(2); + _exit(bounded_listing_ok(bounded_dir, bounded_id) ? 0 : 1); + } + if (pid > 0) + CHECK(waitpid(pid, &status, 0) == pid && WIFEXITED(status) && + WEXITSTATUS(status) == 0); + CHECK(unlink(bounded_path) == 0); + CHECK(rmdir(bounded_dir) == 0); } static void @@ -851,6 +1821,10 @@ main(void) err(1, "mkdtemp"); test_roundtrip(); + test_system_epoch_roundtrip(); + test_epoch_retained_transactions(); + test_epoch_reasons_and_legacy(); + test_epoch_malformed_and_bounds(); test_resume_fidelity(); test_clear(); test_crash_midturn(); blob - 9cc696317e616a67a18200fbffa683dc1c4be6b2 blob + c1798233dfeabe24066d0572937b0b8f974fdb53 --- regress/turn_txn/Makefile +++ regress/turn_txn/Makefile @@ -1,7 +1,8 @@ PROGS= turn_txn_test turn_txn_journal_test SRCS_turn_txn_test= turn_txn_test.c turn_txn.c msg.c log.c xmalloc.c SRCS_turn_txn_journal_test=turn_txn_journal_test.c turn_txn.c journal.c \ - msg.c json.c buf.c log.c xmalloc.c + system_epoch.c compaction.c msg.c json.c buf.c utf8.c \ + log.c xmalloc.c .PATH: ${.CURDIR}/../../src/fugu CFLAGS+= -I${.CURDIR}/../../src/fugu blob - 8cdfe1972919af2a12f240b2233d7d637a1bc04b blob + cccff418f6ed8812dd7341bc393536353fb99589 --- regress/turn_txn/turn_txn_journal_test.c +++ regress/turn_txn/turn_txn_journal_test.c @@ -26,6 +26,7 @@ #include #include "buf.h" +#include "compaction.h" #include "journal.h" #include "log.h" #include "msg.h" @@ -140,6 +141,30 @@ accept_pair(struct turn_txn *txn, const char *user, co } static void +epoch_build(struct system_epoch *epoch, enum system_epoch_reason reason, + const char *summary) +{ + static const char base[] = "test System baseline"; + struct system_epoch_input input; + + memset(&input, 0, sizeof(input)); + input.reason = reason; + input.base.data = base; + input.base.len = sizeof(base) - 1; + input.summary.data = summary; + input.summary.len = summary == NULL ? 0 : strlen(summary); + system_epoch_init(epoch); + CHECK(system_epoch_build(epoch, &input) == 0); +} + +static void +restore_state(const char *id, struct journal_state *state) +{ + journal_state_init(state); + CHECK(journal_restore(dir, id, state) == 0); +} + +static void test_real_journal_terminals(void) { struct job_trace trace = {0}; @@ -175,68 +200,109 @@ test_real_journal_terminals(void) } static void -test_real_journal_resets(void) +test_real_journal_epoch_resets_and_provisional_turns(void) { - static const char personal[] = { 'p', '\0', 'x' }; - static const char project[] = { 'r', '\0', 'y' }; - static const char summary[] = { 's', '\0', 'm' }; - struct turn_txn_context context; - struct turn_txn_jobs jobs; + struct system_epoch start, compact, clear; + struct compaction_files *files, *clear_files; + struct journal_state state; struct job_trace trace = {0}; struct journal *journal; struct turn_txn *txn; - struct msglist replay; - struct buf p, r, compact; + struct turn_txn_projection projection; char id[SESSION_ID_MAX]; txn = session_new(&journal, id, &trace); - accept_pair(txn, "before clear", "old", TURN_TXN_OK); - context.personal = personal; - context.personal_len = sizeof(personal); - context.project = project; - context.project_len = sizeof(project); - turn_txn_clear(txn, &context); - turn_txn_free(txn); - journal_close(journal); - TAILQ_INIT(&replay); - CHECK(journal_replay(dir, id, &replay) == 0); - CHECK(TAILQ_EMPTY(&replay)); - buf_init(&p); - buf_init(&r); - buf_init(&compact); - CHECK(journal_replay_context(dir, id, &p, &r, &compact) == 0); - CHECK(p.len == sizeof(personal) && - memcmp(p.data, personal, sizeof(personal)) == 0); - CHECK(r.len == sizeof(project) && - memcmp(r.data, project, sizeof(project)) == 0); - CHECK(compact.len == 0); - buf_free(&p); - buf_free(&r); - buf_free(&compact); + files = compaction_files_new(); + CHECK(files != NULL); + CHECK(compaction_files_add(files, COMPACTION_FILE_READ, + "README.md", 9) == 0); + epoch_build(&start, SYSTEM_EPOCH_START, NULL); + CHECK(journal_system_epoch(journal, &start, files, NULL, 0) == 0); + journal_sync(journal); + accept_pair(txn, "old user", "old answer", TURN_TXN_OK); - memset(&trace, 0, sizeof(trace)); - TAILQ_INIT(&replay); - journal = journal_create(dir, id, sizeof(id)); - if (journal == NULL) - err(1, "journal_create compact"); - jobs_init(&jobs, &trace); - txn = turn_txn_new(journal, &replay, &jobs); - accept_pair(txn, "before compact", "old", TURN_TXN_OK); - turn_txn_compact(txn, summary, sizeof(summary)); + /* An active epoch is durable but provisional before its terminal. */ + epoch_build(&compact, SYSTEM_EPOCH_COMPACT, "first summary"); + turn_txn_begin(txn); + turn_txn_accept(txn, text_message(ROLE_USER, "new user")); + CHECK(turn_txn_epoch_reset(txn, &compact, files, 2) == 0); + turn_txn_projection(txn, &projection); + CHECK(projection.message_count == 1 && + strcmp(message_text(projection.messages, 0), "new user") == 0); + restore_state(id, &state); + CHECK(state.have_epoch && state.epoch.reason == SYSTEM_EPOCH_START && + message_count(&state.messages) == 2 && + strcmp(message_text(&state.messages, 0), "old user") == 0); + journal_state_free(&state); + + turn_txn_accept(txn, text_message(ROLE_ASSISTANT, "new answer")); + restore_state(id, &state); + CHECK(state.epoch.reason == SYSTEM_EPOCH_START && + message_count(&state.messages) == 2 && + strcmp(message_text(&state.messages, 0), "old user") == 0); + journal_state_free(&state); + turn_txn_commit(txn, TURN_TXN_OK); + restore_state(id, &state); + CHECK(state.have_epoch && state.epoch.reason == SYSTEM_EPOCH_COMPACT && + message_count(&state.messages) == 2 && + strcmp(message_text(&state.messages, 0), "new user") == 0 && + strcmp(message_text(&state.messages, 1), "new answer") == 0); + journal_state_free(&state); + + /* CAP commits the same provisional boundary. */ + system_epoch_free(&compact); + epoch_build(&compact, SYSTEM_EPOCH_COMPACT, "cap summary"); + turn_txn_begin(txn); + turn_txn_accept(txn, text_message(ROLE_USER, "cap user")); + CHECK(turn_txn_epoch_reset(txn, &compact, files, 2) == 0); + turn_txn_accept(txn, text_message(ROLE_ASSISTANT, "cap answer")); + turn_txn_commit(txn, TURN_TXN_CAP); + restore_state(id, &state); + CHECK(state.epoch.reason == SYSTEM_EPOCH_COMPACT && + message_count(&state.messages) == 2 && + strcmp(message_text(&state.messages, 0), "cap user") == 0 && + strcmp(message_text(&state.messages, 1), "cap answer") == 0); + journal_state_free(&state); + + /* Abandon discards the epoch and every pending message around the cut. */ + system_epoch_free(&compact); + epoch_build(&compact, SYSTEM_EPOCH_COMPACT, "abandoned summary"); + turn_txn_begin(txn); + turn_txn_accept(txn, text_message(ROLE_USER, "abandoned user")); + CHECK(turn_txn_epoch_reset(txn, &compact, files, 2) == 0); + turn_txn_accept(txn, text_message(ROLE_ASSISTANT, + "abandoned answer")); + turn_txn_abandon(txn); + turn_txn_projection(txn, &projection); + CHECK(projection.message_count == 2 && + strcmp(message_text(projection.messages, 0), "cap user") == 0); + restore_state(id, &state); + CHECK(message_count(&state.messages) == 2 && + strcmp(message_text(&state.messages, 0), "cap user") == 0 && + strcmp(message_text(&state.messages, 1), "cap answer") == 0); + journal_state_free(&state); + + /* An idle clear is immediate and restores as one atomic boundary. */ + epoch_build(&clear, SYSTEM_EPOCH_CLEAR, NULL); + clear_files = compaction_files_new(); + CHECK(clear_files != NULL); + CHECK(turn_txn_epoch_reset(txn, &clear, files, 2) == -1); + CHECK(turn_txn_epoch_reset(txn, &clear, clear_files, 2) == 0); + restore_state(id, &state); + CHECK(state.have_epoch && state.epoch.reason == SYSTEM_EPOCH_CLEAR && + TAILQ_EMPTY(&state.messages) && + compaction_files_count(state.files, COMPACTION_FILE_READ) == 0 && + compaction_files_count(state.files, + COMPACTION_FILE_MODIFIED) == 0); + journal_state_free(&state); + turn_txn_free(txn); journal_close(journal); - TAILQ_INIT(&replay); - CHECK(journal_replay(dir, id, &replay) == 0); - CHECK(TAILQ_EMPTY(&replay)); - buf_init(&p); - buf_init(&r); - buf_init(&compact); - CHECK(journal_replay_context(dir, id, &p, &r, &compact) == 0); - CHECK(compact.len == sizeof(summary) && - memcmp(compact.data, summary, sizeof(summary)) == 0); - buf_free(&p); - buf_free(&r); - buf_free(&compact); + compaction_files_free(clear_files); + compaction_files_free(files); + system_epoch_free(&clear); + system_epoch_free(&compact); + system_epoch_free(&start); } int @@ -248,7 +314,7 @@ main(void) if (mkdtemp(dir) == NULL) err(1, "mkdtemp"); test_real_journal_terminals(); - test_real_journal_resets(); + test_real_journal_epoch_resets_and_provisional_turns(); (void)snprintf(cmd, sizeof(cmd), "rm -rf %s", dir); (void)system(cmd); REGRESS_END(); blob - d3a5b25b1e3d7c524fd577fc257e6efab6dba794 blob + 5c8e5c78292cca63a073e7afc46bb2905b7338e1 --- regress/turn_txn/turn_txn_test.c +++ regress/turn_txn/turn_txn_test.c @@ -34,9 +34,11 @@ #include #include +#include "compaction.h" #include "log.h" #include "msg.h" #include "journal.h" +#include "system_epoch.h" #include "turn_txn.h" #include "turn_txn_private.h" @@ -53,12 +55,16 @@ struct fixture { size_t message_seen[16]; int sync_calls; size_t sync_seen; - u_char personal[32]; - size_t personal_len; - u_char project[32]; - size_t project_len; - u_char summary[32]; - size_t summary_len; + int epoch_calls; + int epoch_result; + int epoch_turn_n; + enum system_epoch_reason epoch_reason; + size_t epoch_projection_seen; + int epoch_cache_seen; + const struct msglist *expected_retained; + size_t expected_retain_first; + int retained_exact; + int retained_owned; }; static void @@ -142,40 +148,93 @@ journal_sync(struct journal *journal) trace_add(f, "journal.sync"); } -void -journal_clear(struct journal *journal) +static int +block_equal(const struct block *a, const struct block *b) { - trace_add((void *)journal, "journal.clear"); + if (a->type != b->type) + return (0); + switch (a->type) { + case BLOCK_TEXT: + return (a->textlen == b->textlen && (a->textlen == 0 || + memcmp(a->text, b->text, a->textlen) == 0)); + case BLOCK_TOOL_USE: + return (strcmp(a->tool_id, b->tool_id) == 0 && + strcmp(a->tool_name, b->tool_name) == 0 && + a->tool_input_len == b->tool_input_len && + (a->tool_input_len == 0 || + memcmp(a->tool_input, b->tool_input, + a->tool_input_len) == 0)); + case BLOCK_TOOL_RESULT: + return (strcmp(a->tool_use_id, b->tool_use_id) == 0 && + a->result_len == b->result_len && + (a->result_len == 0 || + memcmp(a->result, b->result, a->result_len) == 0) && + a->is_error == b->is_error); + default: + return (0); + } } -void -journal_context(struct journal *journal, const char *personal, - size_t personal_len, - const char *project, size_t project_len) +static int +message_equal(const struct msg *a, const struct msg *b) { - struct fixture *f = (void *)journal; + const struct block *ab, *bb; - f->personal_len = personal_len; - f->project_len = project_len; - if (personal_len <= sizeof(f->personal) && personal_len > 0) - memcpy(f->personal, personal, personal_len); - if (project_len <= sizeof(f->project) && project_len > 0) - memcpy(f->project, project, project_len); - trace_add(f, "journal.context"); + if (a->role != b->role) + return (0); + ab = TAILQ_FIRST(&a->blocks); + bb = TAILQ_FIRST(&b->blocks); + while (ab != NULL && bb != NULL) { + if (!block_equal(ab, bb)) + return (0); + ab = TAILQ_NEXT(ab, entry); + bb = TAILQ_NEXT(bb, entry); + } + return (ab == NULL && bb == NULL); } -void -journal_compact(struct journal *journal, const char *summary, - size_t summary_len) +static int +retained_equal(struct fixture *f, const struct msglist *retained) { - struct fixture *f = (void *)journal; + const struct msg *actual, *expected; + size_t at = 0; - f->summary_len = summary_len; - if (summary_len <= sizeof(f->summary) && summary_len > 0) - memcpy(f->summary, summary, summary_len); - trace_add(f, "journal.compact"); + if (f->expected_retained == NULL) + return (TAILQ_EMPTY(retained)); + expected = TAILQ_FIRST(f->expected_retained); + while (expected != NULL && at++ < f->expected_retain_first) + expected = TAILQ_NEXT(expected, entry); + actual = TAILQ_FIRST(retained); + f->retained_owned = actual == NULL || actual != expected; + while (actual != NULL && expected != NULL) { + if (!message_equal(actual, expected)) + return (0); + actual = TAILQ_NEXT(actual, entry); + expected = TAILQ_NEXT(expected, entry); + } + return (actual == NULL && expected == NULL); } +int +journal_system_epoch(struct journal *journal, + const struct system_epoch *epoch, const struct compaction_files *files, + const struct msglist *retained, int turn_n) +{ + struct fixture *f = (void *)journal; + struct turn_txn_projection p; + + (void)files; + projection(f, &p); + f->epoch_calls++; + f->epoch_turn_n = turn_n; + f->epoch_reason = epoch->reason; + f->epoch_projection_seen = p.message_count; + f->epoch_cache_seen = p.cache_tail_msg; + f->retained_exact = retained_equal(f, retained); + trace_add(f, "journal.epoch"); + return (f->epoch_result); +} + static void jobs_begin(void *arg) { @@ -264,6 +323,39 @@ first_text(const struct msglist *messages) return (b->text); } +static struct msg * +message_at(const struct msglist *messages, size_t index) +{ + struct msg *message; + size_t at = 0; + + TAILQ_FOREACH(message, messages, entry) + if (at++ == index) + return (message); + return (NULL); +} + +static const char * +text_at(const struct msglist *messages, size_t index) +{ + struct msg *message; + struct block *block; + + message = message_at(messages, index); + if (message == NULL || (block = TAILQ_FIRST(&message->blocks)) == NULL || + block->type != BLOCK_TEXT) + return (NULL); + return (block->text); +} + +static void +fake_epoch(struct system_epoch *epoch, enum system_epoch_reason reason) +{ + memset(epoch, 0, sizeof(*epoch)); + epoch->version = SYSTEM_EPOCH_VERSION; + epoch->reason = reason; +} + static int fatal_projection_sent(struct turn_txn *txn, const struct turn_txn_projection *projection) @@ -497,68 +589,316 @@ test_cap_does_not_claim_unsent_messages(void) } static void -test_resume_clear_compact_and_cache_reset_ownership(void) +test_idle_epoch_reset_clones_exact_retained_suffix(void) { - static const char personal[] = { 'p', '\0', 'x' }; - static const char project[] = { 'r', '\0', 'y', 'z' }; - static const char summary[] = { 's', '\0', 'm' }; - struct turn_txn_context context; + static const char text[] = { 'a', '\0', 'b' }; + static const char input[] = { '{', '"', 'x', '"', ':', '"', '\0', + '"', '}' }; + static const char result[] = { 'r', '\0', 'x' }; + struct system_epoch epoch; struct fixture f; struct msglist seed; - struct turn_txn_projection p; + struct msg *message; + struct block *block; + struct turn_txn_projection before, after; TAILQ_INIT(&seed); - append_text(&seed, ROLE_USER, "resume one"); - append_text(&seed, ROLE_ASSISTANT, "resume two"); + append_text(&seed, ROLE_USER, "discard"); + message = msg_new(ROLE_ASSISTANT); + msg_add_text(message, text, sizeof(text)); + msg_add_tool_use(message, "call-1", "read", input, sizeof(input)); + block = msg_add_text(message, NULL, 0); + free(block->text); + block->text = NULL; + block = msg_add_tool_use(message, "empty-call", "read", NULL, 0); + free(block->tool_input); + block->tool_input = NULL; + TAILQ_INSERT_TAIL(&seed, message, entry); + message = msg_new(ROLE_USER); + msg_add_tool_result(message, "call-1", result, sizeof(result), 1); + block = msg_add_tool_result(message, "empty-call", NULL, 0, 0); + free(block->result); + block->result = NULL; + TAILQ_INSERT_TAIL(&seed, message, entry); fixture_new(&f, &seed); - CHECK(TAILQ_EMPTY(&seed)); - projection(&f, &p); - CHECK(p.message_count == 2 && p.cache_tail_msg == -1 && - strcmp(first_text(p.messages), "resume one") == 0); + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); + projection(&f, &before); + f.expected_retained = before.messages; + f.expected_retain_first = 1; + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 1) == 0); + CHECK(strcmp(f.trace, "journal.epoch,journal.sync") == 0); + CHECK(f.epoch_calls == 1 && f.epoch_turn_n == 0 && + f.epoch_reason == SYSTEM_EPOCH_COMPACT && + f.epoch_projection_seen == 3 && f.epoch_cache_seen == -1); + CHECK(f.retained_exact && f.retained_owned && f.sync_seen == 3); + projection(&f, &after); + CHECK(after.message_count == 2 && after.cache_tail_msg == -1 && + after.token != before.token); + message = message_at(after.messages, 0); + block = message == NULL ? NULL : TAILQ_FIRST(&message->blocks); + CHECK(message != NULL && message->role == ROLE_ASSISTANT && + block != NULL && block->type == BLOCK_TEXT && + block->textlen == sizeof(text) && + memcmp(block->text, text, sizeof(text)) == 0); + block = block == NULL ? NULL : TAILQ_NEXT(block, entry); + CHECK(block != NULL && block->type == BLOCK_TOOL_USE && + strcmp(block->tool_id, "call-1") == 0 && + strcmp(block->tool_name, "read") == 0 && + block->tool_input_len == sizeof(input) && + memcmp(block->tool_input, input, sizeof(input)) == 0); + block = TAILQ_NEXT(block, entry); + CHECK(block != NULL && block->type == BLOCK_TEXT && + block->textlen == 0 && block->text != NULL); + block = TAILQ_NEXT(block, entry); + CHECK(block != NULL && block->type == BLOCK_TOOL_USE && + block->tool_input_len == 0 && block->tool_input != NULL); + message = message_at(after.messages, 1); + block = message == NULL ? NULL : TAILQ_FIRST(&message->blocks); + CHECK(message != NULL && message->role == ROLE_USER && block != NULL && + block->type == BLOCK_TOOL_RESULT && block->result_len == sizeof(result) + && memcmp(block->result, result, sizeof(result)) == 0 && + block->is_error == 1); + block = TAILQ_NEXT(block, entry); + CHECK(block != NULL && block->type == BLOCK_TOOL_RESULT && + block->result_len == 0 && block->result != NULL && + block->is_error == 0); + turn_txn_free(f.txn); +} + +static void +test_epoch_codec_failure_preserves_projection_and_cursor(void) +{ + struct system_epoch epoch; + struct fixture f; + struct msglist seed; + struct msg *first, *second; + struct turn_txn_projection before, after; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "one"); + append_text(&seed, ROLE_ASSISTANT, "two"); + fixture_new(&f, &seed); turn_txn_begin(f.txn); - turn_txn_accept(f.txn, text_message(ROLE_USER, "sent")); - projection(&f, &p); - turn_txn_projection_sent(f.txn, &p); - turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "reply")); + projection(&f, &before); + turn_txn_projection_sent(f.txn, &before); turn_txn_commit(f.txn, TURN_TXN_OK); - projection(&f, &p); - CHECK(p.message_count == 4 && p.cache_tail_msg == 2); - turn_txn_cache_reset(f.txn); - projection(&f, &p); - CHECK(p.message_count == 4 && p.cache_tail_msg == -1); - - context.personal = personal; - context.personal_len = sizeof(personal); - context.project = project; - context.project_len = sizeof(project); + projection(&f, &before); + first = message_at(before.messages, 0); + second = message_at(before.messages, 1); + CHECK(before.cache_tail_msg == 1); + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); + f.expected_retained = before.messages; + f.expected_retain_first = 1; + f.epoch_result = -1; trace_reset(&f); - turn_txn_clear(f.txn, &context); - CHECK(strcmp(f.trace, "journal.clear,journal.context,journal.sync") == 0); - CHECK(f.sync_seen == 4); - CHECK(f.personal_len == sizeof(personal) && - memcmp(f.personal, personal, sizeof(personal)) == 0); - CHECK(f.project_len == sizeof(project) && - memcmp(f.project, project, sizeof(project)) == 0); + + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 1) == -1); + CHECK(strcmp(f.trace, "journal.epoch") == 0 && f.sync_calls == 1); + projection(&f, &after); + CHECK(after.token == before.token && after.message_count == 2 && + after.cache_tail_msg == 1 && message_at(after.messages, 0) == first && + message_at(after.messages, 1) == second); + CHECK(strcmp(text_at(after.messages, 0), "one") == 0 && + strcmp(text_at(after.messages, 1), "two") == 0); + turn_txn_free(f.txn); +} + +static void +test_idle_clear_is_durable_before_empty_projection(void) +{ + struct system_epoch epoch; + struct fixture f; + struct msglist seed; + struct turn_txn_projection p; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "one"); + append_text(&seed, ROLE_ASSISTANT, "two"); + fixture_new(&f, &seed); + fake_epoch(&epoch, SYSTEM_EPOCH_CLEAR); + f.expected_retained = NULL; + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 2) == 0); + CHECK(strcmp(f.trace, "journal.epoch,journal.sync") == 0 && + f.epoch_projection_seen == 2 && f.sync_seen == 2 && + f.retained_exact); projection(&f, &p); CHECK(p.message_count == 0 && p.cache_tail_msg == -1); + turn_txn_free(f.txn); +} +static void +test_active_epoch_cut_commits_and_cools_cache(void) +{ + struct system_epoch epoch; + struct fixture f; + struct msglist seed; + struct turn_txn_projection sent, before, after; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "old"); + append_text(&seed, ROLE_ASSISTANT, "keep"); + fixture_new(&f, &seed); + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); turn_txn_begin(f.txn); - turn_txn_accept(f.txn, text_message(ROLE_USER, "after clear")); - projection(&f, &p); - turn_txn_projection_sent(f.txn, &p); - turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "after reply")); - turn_txn_commit(f.txn, TURN_TXN_OK); + turn_txn_accept(f.txn, text_message(ROLE_USER, "question")); + projection(&f, &sent); + turn_txn_projection_sent(f.txn, &sent); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "answer")); + projection(&f, &before); + f.expected_retained = before.messages; + f.expected_retain_first = 1; trace_reset(&f); - turn_txn_compact(f.txn, summary, sizeof(summary)); - CHECK(strcmp(f.trace, "journal.compact,journal.sync") == 0); - CHECK(f.sync_seen == 2 && f.summary_len == sizeof(summary) && - memcmp(f.summary, summary, sizeof(summary)) == 0); + + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 1) == 0); + CHECK(strcmp(f.trace, "journal.epoch,journal.sync") == 0 && + f.epoch_turn_n == 1 && f.epoch_projection_seen == 4 && + f.epoch_cache_seen == 2 && f.retained_exact && f.retained_owned); + projection(&f, &after); + CHECK(after.message_count == 3 && after.cache_tail_msg == -1 && + after.token != before.token); + CHECK(fatal_projection_sent(f.txn, &before)); + turn_txn_accept(f.txn, text_message(ROLE_USER, "after cut")); + turn_txn_commit(f.txn, TURN_TXN_OK); + projection(&f, &after); + CHECK(after.message_count == 4 && + strcmp(text_at(after.messages, 0), "keep") == 0 && + strcmp(text_at(after.messages, 1), "question") == 0 && + strcmp(text_at(after.messages, 2), "answer") == 0 && + strcmp(text_at(after.messages, 3), "after cut") == 0); + + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "drop")); + turn_txn_abandon(f.txn); + projection(&f, &after); + CHECK(after.message_count == 4 && + strcmp(text_at(after.messages, 0), "keep") == 0); + turn_txn_free(f.txn); +} + +static void +test_active_cut_through_pending_first_abandons_to_preturn_projection(void) +{ + struct system_epoch epoch; + struct fixture f; + struct msglist seed; + struct turn_txn_projection before, after; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "committed user"); + append_text(&seed, ROLE_ASSISTANT, "committed answer"); + fixture_new(&f, &seed); + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "pending first")); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "pending second")); + projection(&f, &before); + f.expected_retained = before.messages; + f.expected_retain_first = 3; + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 3) == 0); + projection(&f, &after); + CHECK(after.message_count == 1 && + strcmp(first_text(after.messages), "pending second") == 0); + turn_txn_accept(f.txn, text_message(ROLE_USER, "post-cut pending")); + turn_txn_abandon(f.txn); + projection(&f, &after); + CHECK(after.message_count == 2 && after.cache_tail_msg == -1 && + strcmp(text_at(after.messages, 0), "committed user") == 0 && + strcmp(text_at(after.messages, 1), "committed answer") == 0); + turn_txn_free(f.txn); +} + +static void +test_multiple_active_cuts_keep_original_rollback_snapshot(void) +{ + struct system_epoch epoch; + struct fixture f; + struct msglist seed; + struct turn_txn_projection p; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "base one"); + append_text(&seed, ROLE_ASSISTANT, "base two"); + append_text(&seed, ROLE_USER, "base three"); + fixture_new(&f, &seed); + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "pending one")); + turn_txn_accept(f.txn, text_message(ROLE_USER, "pending two")); projection(&f, &p); - CHECK(p.message_count == 0 && p.cache_tail_msg == -1); + f.expected_retained = p.messages; + f.expected_retain_first = 2; + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 2) == 0); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "pending three")); + projection(&f, &p); + f.expected_retained = p.messages; + f.expected_retain_first = 2; + CHECK(turn_txn_epoch_reset(f.txn, &epoch, + (const struct compaction_files *)(const void *)&f, 2) == 0); + CHECK(f.epoch_calls == 2 && f.epoch_turn_n == 1 && f.retained_exact); + projection(&f, &p); + CHECK(p.message_count == 2 && + strcmp(text_at(p.messages, 0), "pending two") == 0 && + strcmp(text_at(p.messages, 1), "pending three") == 0); + turn_txn_abandon(f.txn); + projection(&f, &p); + CHECK(p.message_count == 3 && + strcmp(text_at(p.messages, 0), "base one") == 0 && + strcmp(text_at(p.messages, 1), "base two") == 0 && + strcmp(text_at(p.messages, 2), "base three") == 0); turn_txn_free(f.txn); } +static void +test_epoch_reset_validation_is_fail_closed(void) +{ + struct system_epoch epoch; + struct fixture f; + struct msglist seed; + struct turn_txn_projection before, after; + const struct compaction_files *files; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "one"); + append_text(&seed, ROLE_ASSISTANT, "two"); + fixture_new(&f, &seed); + files = (const struct compaction_files *)(const void *)&f; + projection(&f, &before); + fake_epoch(&epoch, SYSTEM_EPOCH_START); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 2) == -1); + fake_epoch(&epoch, SYSTEM_EPOCH_MIGRATION); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 2) == -1); + fake_epoch(&epoch, (enum system_epoch_reason)99); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 2) == -1); + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); + CHECK(turn_txn_epoch_reset(f.txn, NULL, files, 2) == -1); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, NULL, 2) == -1); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 3) == -1); + fake_epoch(&epoch, SYSTEM_EPOCH_CLEAR); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 1) == -1); + epoch.summary.data = "not empty"; + epoch.summary.len = 9; + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 2) == -1); + epoch.summary.data = NULL; + epoch.summary.len = 0; + projection(&f, &after); + CHECK(f.epoch_calls == 0 && after.token == before.token && + after.message_count == before.message_count && + message_at(after.messages, 0) == message_at(before.messages, 0)); + + turn_txn_begin(f.txn); + CHECK(turn_txn_epoch_reset(f.txn, &epoch, files, 2) == -1); + CHECK(f.epoch_calls == 0); + turn_txn_abandon(f.txn); + turn_txn_free(f.txn); +} + enum fatal_case { FATAL_BEGIN_ACTIVE, FATAL_ACCEPT_IDLE, @@ -570,10 +910,7 @@ enum fatal_case { FATAL_COMMIT_OUTCOME, FATAL_ABANDON_IDLE, FATAL_RESET_NULL, - FATAL_CLEAR_ACTIVE, - FATAL_CLEAR_CONTEXT, - FATAL_COMPACT_ACTIVE, - FATAL_COMPACT_NULL, + FATAL_EPOCH_RESET_NULL, FATAL_FREE_ACTIVE, FATAL_NEW_NULL_SEED, FATAL_NEW_NULL_JOURNAL, @@ -584,7 +921,7 @@ static void run_fatal_case(enum fatal_case which) { struct turn_txn_jobs jobs; - struct turn_txn_context context; + struct system_epoch epoch; struct turn_txn_projection p; struct fixture f; struct msglist seed; @@ -642,22 +979,11 @@ run_fatal_case(enum fatal_case which) case FATAL_RESET_NULL: turn_txn_cache_reset(NULL); break; - case FATAL_CLEAR_ACTIVE: - turn_txn_begin(f.txn); - turn_txn_clear(f.txn, NULL); + case FATAL_EPOCH_RESET_NULL: + fake_epoch(&epoch, SYSTEM_EPOCH_COMPACT); + (void)turn_txn_epoch_reset(NULL, &epoch, + (const struct compaction_files *)(const void *)&f, 0); break; - case FATAL_CLEAR_CONTEXT: - memset(&context, 0, sizeof(context)); - context.personal_len = 1; - turn_txn_clear(f.txn, &context); - break; - case FATAL_COMPACT_ACTIVE: - turn_txn_begin(f.txn); - turn_txn_compact(f.txn, "x", 1); - break; - case FATAL_COMPACT_NULL: - turn_txn_compact(f.txn, NULL, 1); - break; case FATAL_FREE_ACTIVE: turn_txn_begin(f.txn); turn_txn_free(f.txn); @@ -708,7 +1034,13 @@ main(void) test_active_cache_reset_cools_cursor_without_ending_turn(); test_foreign_snapshot_is_rejected(); test_cap_does_not_claim_unsent_messages(); - test_resume_clear_compact_and_cache_reset_ownership(); + test_idle_epoch_reset_clones_exact_retained_suffix(); + test_epoch_codec_failure_preserves_projection_and_cursor(); + test_idle_clear_is_durable_before_empty_projection(); + test_active_epoch_cut_commits_and_cools_cache(); + test_active_cut_through_pending_first_abandons_to_preturn_projection(); + test_multiple_active_cuts_keep_original_rollback_snapshot(); + test_epoch_reset_validation_is_fail_closed(); test_contract_failures_are_fail_closed(); REGRESS_END(); } blob - 5084f1bca3752b28da8dee039a5313fa4b8e35cd blob + 0c567c4088e25bf7ef84c288e1837ec1e06faaf1 --- src/fugu/journal.c +++ src/fugu/journal.c @@ -35,28 +35,124 @@ #include "buf.h" #include "json.h" #include "msg.h" +#include "compaction.h" +#include "system_epoch.h" #include "journal.h" #define JOURNAL_VER 1 -#define JOURNAL_RECORD_MAX (32 * 1024 * 1024) #define JOURNAL_READ_CHUNK (64 * 1024) +#define JOURNAL_CRASH_TAIL "\t#fugu-crash-tail-v1" +#define JOURNAL_CRASH_TAIL_LEN (sizeof(JOURNAL_CRASH_TAIL) - 1) +#define JOURNAL_READ_MAX (JOURNAL_RECORD_MAX + JOURNAL_CRASH_TAIL_LEN) #define RECORD_ERROR (-2) #define RECORD_OVERSIZE (-1) +#define RECORD_STRUCTURED_OVERSIZE (-3) +#define RECORD_SEALED_TAIL (-4) struct journal { int fd; + int turn_n; char id[SESSION_ID_MAX]; char path[1024]; }; +void +journal_state_init(struct journal_state *state) +{ + memset(state, 0, sizeof(*state)); + TAILQ_INIT(&state->messages); + system_epoch_init(&state->epoch); + state->files = compaction_files_new(); + buf_init(&state->legacy_personal); + buf_init(&state->legacy_project); + buf_init(&state->legacy_summary); +} + +void +journal_state_free(struct journal_state *state) +{ + if (state == NULL) + return; + msglist_free(&state->messages); + system_epoch_free(&state->epoch); + compaction_files_free(state->files); + buf_free(&state->legacy_personal); + buf_free(&state->legacy_project); + buf_free(&state->legacy_summary); + memset(state, 0, sizeof(*state)); +} + struct record_reader { int fd; struct buf in; int eof; int dropping; + int drop_epoch; + size_t epoch_match; + size_t seal_match; }; +static int +line_mentions_epoch(const void *data, size_t len) +{ + static const char marker[] = "\"system-epoch\""; + const u_char *p = data; + size_t i; + + if (len < sizeof(marker) - 1) + return (0); + for (i = 0; i <= len - (sizeof(marker) - 1); i++) + if (memcmp(p + i, marker, sizeof(marker) - 1) == 0) + return (1); + return (0); +} + +static int +line_has_crash_tail(const void *data, size_t len) +{ + return (len >= JOURNAL_CRASH_TAIL_LEN && + memcmp((const u_char *)data + len - JOURNAL_CRASH_TAIL_LEN, + JOURNAL_CRASH_TAIL, JOURNAL_CRASH_TAIL_LEN) == 0); +} + static void +record_drop_scan(struct record_reader *r, const void *data, size_t len) +{ + static const char epoch_marker[] = "\"system-epoch\""; + static const char seal_marker[] = JOURNAL_CRASH_TAIL; + const u_char *p = data; + size_t i; + + for (i = 0; i < len; i++) { + if (!r->drop_epoch) { + if (p[i] == (u_char)epoch_marker[r->epoch_match]) { + r->epoch_match++; + if (r->epoch_match == sizeof(epoch_marker) - 1) { + r->drop_epoch = 1; + r->epoch_match = 0; + } + } else + r->epoch_match = p[i] == + (u_char)epoch_marker[0] ? 1 : 0; + } + if (r->seal_match == sizeof(seal_marker) - 1) + r->seal_match = 0; + if (p[i] == (u_char)seal_marker[r->seal_match]) { + r->seal_match++; + } else + r->seal_match = p[i] == (u_char)seal_marker[0] ? 1 : 0; + } +} + +static void +record_drop_reset(struct record_reader *r) +{ + r->drop_epoch = 0; + r->epoch_match = 0; + r->seal_match = 0; +} + +static void record_reader_init(struct record_reader *r, int fd) { memset(r, 0, sizeof(*r)); @@ -72,8 +168,8 @@ record_reader_done(struct record_reader *r) /* * Read one newline-terminated NDJSON record without imposing a whole-file - * ceiling. Returns 1 with a record, -1 for one oversized record (already - * discarded), 0 at EOF; an unterminated final fragment is ignored. + * ceiling. Returns 1 with a record, a RECORD_* classification for a skipped + * record, or 0 at EOF; an unterminated final fragment is ignored. */ static int record_next(struct record_reader *r, struct buf *line) @@ -88,16 +184,39 @@ record_next(struct record_reader *r, struct buf *line) (nl = memchr(r->in.data, '\n', r->in.len)) != NULL) { len = (size_t)(nl - r->in.data); - if (len <= JOURNAL_RECORD_MAX && len > 0) + if (len <= JOURNAL_READ_MAX && len > 0) buf_add(line, r->in.data, len); + if (len > JOURNAL_READ_MAX) { + record_drop_reset(r); + record_drop_scan(r, r->in.data, len); + } buf_drain(&r->in, len + 1); - return (len > JOURNAL_RECORD_MAX ? -1 : 1); + if (len <= JOURNAL_READ_MAX && + line_has_crash_tail(line->data, line->len)) + return (RECORD_SEALED_TAIL); + if (len <= JOURNAL_RECORD_MAX) + return (1); + if (len > JOURNAL_READ_MAX && + r->seal_match == JOURNAL_CRASH_TAIL_LEN) { + record_drop_reset(r); + return (RECORD_SEALED_TAIL); + } + if (len > JOURNAL_READ_MAX) { + len = r->drop_epoch; + record_drop_reset(r); + return (len ? RECORD_STRUCTURED_OVERSIZE : + RECORD_OVERSIZE); + } + return (line_mentions_epoch(line->data, line->len) ? + RECORD_STRUCTURED_OVERSIZE : RECORD_OVERSIZE); } if (r->eof) { buf_reset(&r->in); return (0); } - if (!r->dropping && r->in.len > JOURNAL_RECORD_MAX) { + if (!r->dropping && r->in.len > JOURNAL_READ_MAX) { + record_drop_reset(r); + record_drop_scan(r, r->in.data, r->in.len); buf_reset(&r->in); r->dropping = 1; } @@ -113,13 +232,21 @@ record_next(struct record_reader *r, struct buf *line) } if (r->dropping) { nl = memchr(tmp, '\n', (size_t)n); + len = nl == NULL ? (size_t)n : (size_t)(nl - tmp); + record_drop_scan(r, tmp, len); if (nl == NULL) continue; rest = (size_t)n - (size_t)(nl - tmp) - 1; if (rest > 0) buf_add(&r->in, nl + 1, rest); r->dropping = 0; - return (-1); + if (r->seal_match == JOURNAL_CRASH_TAIL_LEN) { + record_drop_reset(r); + return (RECORD_SEALED_TAIL); + } + len = r->drop_epoch; + record_drop_reset(r); + return (len ? RECORD_STRUCTURED_OVERSIZE : RECORD_OVERSIZE); } buf_add(&r->in, tmp, (size_t)n); } @@ -228,6 +355,10 @@ struct journal * journal_open(const char *dir, const char *id) { struct journal *j; + struct stat st; + char last; + ssize_t n; + int fd, partial = 0; if (!valid_id(id)) return (NULL); @@ -236,11 +367,45 @@ journal_open(const char *dir, const char *id) if ((size_t)snprintf(j->path, sizeof(j->path), "%s/%s.ndjson", dir, id) >= sizeof(j->path)) fatalx("journal path too long"); + if ((fd = open(j->path, O_RDONLY)) == -1) { + free(j); + return (NULL); + } + if (fstat(fd, &st) == -1) { + close(fd); + free(j); + return (NULL); + } + if (st.st_size > 0) { + do { + n = pread(fd, &last, 1, st.st_size - 1); + } while (n == -1 && errno == EINTR); + if (n != 1) { + close(fd); + free(j); + return (NULL); + } + partial = last != '\n'; + } + close(fd); j->fd = open(j->path, O_WRONLY|O_APPEND); if (j->fd == -1) { free(j); return (NULL); } + /* + * Never let a retry complete the ignored crash fragment. The sentinel + * is appended without truncating history, and its newline is the final + * byte of the write so another crash still leaves an ignorable tail. + */ + if (partial) { + static const char seal[] = JOURNAL_CRASH_TAIL "\n"; + + if (write(j->fd, seal, sizeof(seal) - 1) != + (ssize_t)(sizeof(seal) - 1)) + fatal("journal crash-tail seal"); + journal_sync(j); + } return (j); } @@ -278,12 +443,452 @@ journal_sync(struct journal *j) } } +static const char * +epoch_reason_name(enum system_epoch_reason reason) +{ + switch (reason) { + case SYSTEM_EPOCH_START: + return ("start"); + case SYSTEM_EPOCH_CLEAR: + return ("clear"); + case SYSTEM_EPOCH_COMPACT: + return ("compact"); + case SYSTEM_EPOCH_MIGRATION: + return ("migration"); + } + return (NULL); +} + +static void +json_hex_span(struct json_out *jo, const void *data, size_t len) +{ + static const char digits[] = "0123456789abcdef"; + const u_char *p = data; + struct buf encoded; + size_t i; + + buf_init(&encoded); + for (i = 0; i < len; i++) { + buf_addc(&encoded, digits[p[i] >> 4]); + buf_addc(&encoded, digits[p[i] & 0x0f]); + } + json_str(jo, (const char *)encoded.data, encoded.len); + buf_free(&encoded); +} + +static void +json_hex(struct json_out *jo, const struct buf *span) +{ + json_hex_span(jo, span->data, span->len); +} + +static size_t +message_tool_results(const struct msg *m) +{ + const struct block *blk; + size_t count = 0; + + if (m == NULL || m->role != ROLE_USER) + return (0); + TAILQ_FOREACH(blk, &m->blocks, entry) + if (blk->type == BLOCK_TOOL_RESULT) + count++; + return (count); +} + +static int +assistant_results_complete(const struct msg *assistant, + const struct msg *results) +{ + const struct block *call, *result; + size_t calls = 0, matches, nresults; + + TAILQ_FOREACH(call, &assistant->blocks, entry) + if (call->type == BLOCK_TOOL_USE) + calls++; + nresults = message_tool_results(results); + if (calls == 0) + return (nresults == 0); + if (results == NULL || results->role != ROLE_USER || calls != nresults) + return (0); + TAILQ_FOREACH(call, &assistant->blocks, entry) { + if (call->type != BLOCK_TOOL_USE) + continue; + matches = 0; + TAILQ_FOREACH(result, &results->blocks, entry) + if (result->type == BLOCK_TOOL_RESULT && + strcmp(call->tool_id, result->tool_use_id) == 0) + matches++; + if (matches != 1) + return (0); + } + return (1); +} + +static int +retained_projection_valid(const struct msglist *retained) +{ + const struct msg *m, *previous = NULL; + const struct block *blk; + + if (retained == NULL) + return (1); + TAILQ_FOREACH(m, retained, entry) { + if ((m->role != ROLE_USER && m->role != ROLE_ASSISTANT) || + TAILQ_EMPTY(&m->blocks) || + (previous != NULL && previous->role == m->role)) + return (0); + TAILQ_FOREACH(blk, &m->blocks, entry) { + switch (blk->type) { + case BLOCK_TEXT: + if (blk->text == NULL && blk->textlen > 0) + return (0); + break; + case BLOCK_TOOL_USE: + if (m->role != ROLE_ASSISTANT || + blk->tool_id == NULL || blk->tool_id[0] == '\0' || + blk->tool_name == NULL || + blk->tool_name[0] == '\0' || + (blk->tool_input == NULL && + blk->tool_input_len > 0) || + !json_valid_object(blk->tool_input, + blk->tool_input_len)) + return (0); + break; + case BLOCK_TOOL_RESULT: + if (m->role != ROLE_USER || + blk->tool_use_id == NULL || + blk->tool_use_id[0] == '\0' || + (blk->result == NULL && blk->result_len > 0) || + (blk->is_error != 0 && blk->is_error != 1)) + return (0); + break; + default: + return (0); + } + } + if (m->role == ROLE_USER && message_tool_results(m) > 0 && + previous == NULL) + return (0); + if (previous != NULL && previous->role == ROLE_ASSISTANT && + !assistant_results_complete(previous, m)) + return (0); + previous = m; + } + if (previous != NULL && previous->role == ROLE_ASSISTANT && + !assistant_results_complete(previous, NULL)) + return (0); + return (1); +} + +static int +preflight_add(size_t *total, size_t len, size_t width, size_t overhead) +{ + size_t add; + + if (overhead > JOURNAL_RECORD_MAX || + len > (JOURNAL_RECORD_MAX - overhead) / width) + return (-1); + add = len * width + overhead; + if (*total > JOURNAL_RECORD_MAX - add) + return (-1); + *total += add; + return (0); +} + +static int +preflight_cstr(size_t *total, const char *s) +{ + size_t len; + + if (s == NULL) + return (-1); + len = strnlen(s, JOURNAL_RECORD_MAX + 1UL); + if (len > JOURNAL_RECORD_MAX) + return (-1); + return (preflight_add(total, len, 2, 0)); +} + +static int +epoch_record_preflight(const struct system_epoch *epoch, + const struct compaction_files *files, const struct msglist *retained) +{ + const struct buf *span; + const struct msg *m; + const struct block *blk; + const void *path; + enum compaction_file_kind kind; + size_t i, len, n, total = 512; + + span = &epoch->base; + if ((span->data == NULL && span->len > 0) || + preflight_add(&total, span->len, 2, 0) == -1) + return (-1); + span = &epoch->personal; + if ((span->data == NULL && span->len > 0) || + preflight_add(&total, span->len, 2, 0) == -1) + return (-1); + span = &epoch->project; + if ((span->data == NULL && span->len > 0) || + preflight_add(&total, span->len, 2, 0) == -1) + return (-1); + span = &epoch->summary; + if ((span->data == NULL && span->len > 0) || + preflight_add(&total, span->len, 2, 0) == -1) + return (-1); + span = &epoch->full; + if ((span->data == NULL && span->len > 0) || + preflight_add(&total, span->len, 2, 0) == -1) + return (-1); + span = &epoch->restricted; + if ((span->data == NULL && span->len > 0) || + preflight_add(&total, span->len, 2, 0) == -1) + return (-1); + + for (kind = COMPACTION_FILE_READ; + kind <= COMPACTION_FILE_MODIFIED; kind++) { + n = files == NULL ? 0 : compaction_files_count(files, kind); + if (n > COMPACTION_FILE_COUNT_MAX) + return (-1); + for (i = 0; i < n; i++) { + if (compaction_files_get(files, kind, i, &path, &len) == -1 || + (path == NULL && len > 0) || + preflight_add(&total, len, 6, 8) == -1) + return (-1); + } + } + if (retained == NULL) + return (0); + TAILQ_FOREACH(m, retained, entry) { + if (preflight_add(&total, 0, 1, 64) == -1) + return (-1); + TAILQ_FOREACH(blk, &m->blocks, entry) { + if (preflight_add(&total, 0, 1, 128) == -1) + return (-1); + switch (blk->type) { + case BLOCK_TEXT: + if ((blk->text == NULL && blk->textlen > 0) || + preflight_add(&total, blk->textlen, 2, 0) == -1) + return (-1); + break; + case BLOCK_TOOL_USE: + if (preflight_cstr(&total, blk->tool_id) == -1 || + preflight_cstr(&total, blk->tool_name) == -1 || + (blk->tool_input == NULL && + blk->tool_input_len > 0) || + preflight_add(&total, blk->tool_input_len, + 2, 0) == -1) + return (-1); + break; + case BLOCK_TOOL_RESULT: + if (preflight_cstr(&total, blk->tool_use_id) == -1 || + (blk->result == NULL && blk->result_len > 0) || + preflight_add(&total, blk->result_len, 2, 0) == -1) + return (-1); + break; + default: + return (-1); + } + } + } + return (0); +} + +static int +emit_file_array(struct json_out *jo, const struct compaction_files *files, + enum compaction_file_kind kind) +{ + const void *path; + size_t i, len, n; + + n = files == NULL ? 0 : compaction_files_count(files, kind); + json_arr_begin(jo); + for (i = 0; i < n; i++) { + if (compaction_files_get(files, kind, i, &path, &len) == -1) + return (-1); + json_str(jo, path, len); + } + json_arr_end(jo); + return (0); +} + +static int +emit_retained(struct json_out *jo, const struct msglist *retained, + size_t *countp) +{ + const struct msg *m; + const struct block *blk; + size_t count = 0; + + json_arr_begin(jo); + if (retained != NULL) { + TAILQ_FOREACH(m, retained, entry) { + if (m->role != ROLE_USER && m->role != ROLE_ASSISTANT) + return (-1); + if (TAILQ_EMPTY(&m->blocks)) + return (-1); + json_obj_begin(jo); + json_key(jo, "role"); + json_cstr(jo, m->role == ROLE_USER ? "user" : + "assistant"); + json_key(jo, "blocks"); + json_arr_begin(jo); + TAILQ_FOREACH(blk, &m->blocks, entry) { + json_obj_begin(jo); + switch (blk->type) { + case BLOCK_TEXT: + if (blk->text == NULL && blk->textlen > 0) + return (-1); + json_key(jo, "type"); + json_cstr(jo, "text"); + json_key(jo, "text"); + json_hex_span(jo, blk->text, blk->textlen); + break; + case BLOCK_TOOL_USE: + if (m->role != ROLE_ASSISTANT || + blk->tool_id == NULL || blk->tool_id[0] == '\0' || + blk->tool_name == NULL || + blk->tool_name[0] == '\0' || + (blk->tool_input == NULL && + blk->tool_input_len > 0) || + !json_valid_object(blk->tool_input, + blk->tool_input_len)) + return (-1); + json_key(jo, "type"); + json_cstr(jo, "tool-use"); + json_key(jo, "id"); + json_hex_span(jo, blk->tool_id, + strlen(blk->tool_id)); + json_key(jo, "name"); + json_hex_span(jo, blk->tool_name, + strlen(blk->tool_name)); + json_key(jo, "input"); + json_hex_span(jo, blk->tool_input, + blk->tool_input_len); + break; + case BLOCK_TOOL_RESULT: + if (m->role != ROLE_USER || + blk->tool_use_id == NULL || + blk->tool_use_id[0] == '\0' || + (blk->result == NULL && blk->result_len > 0) || + (blk->is_error != 0 && blk->is_error != 1)) + return (-1); + json_key(jo, "type"); + json_cstr(jo, "tool-result"); + json_key(jo, "tool_use_id"); + json_hex_span(jo, blk->tool_use_id, + strlen(blk->tool_use_id)); + json_key(jo, "content"); + json_hex_span(jo, blk->result, + blk->result_len); + json_key(jo, "is_error"); + json_bool(jo, blk->is_error); + break; + default: + return (-1); + } + json_obj_end(jo); + } + json_arr_end(jo); + json_obj_end(jo); + count++; + } + } + json_arr_end(jo); + *countp = count; + return (0); +} + +int +journal_system_epoch(struct journal *j, const struct system_epoch *epoch, + const struct compaction_files *files, const struct msglist *retained, + int turn_n) +{ + struct system_epoch checked; + struct buf b; + struct json parsed; + struct json_out jo; + const char *reason; + size_t retained_count; + int ok = -1; + + if (j == NULL || epoch == NULL || + (reason = epoch_reason_name(epoch->reason)) == NULL || turn_n < 0 || + (turn_n > 0 && epoch->reason != SYSTEM_EPOCH_COMPACT) || + turn_n != j->turn_n || + (retained != NULL && !TAILQ_EMPTY(retained) && + epoch->reason != SYSTEM_EPOCH_COMPACT) || + (epoch->reason == SYSTEM_EPOCH_CLEAR && + (epoch->summary.len != 0 || + (files != NULL && + (compaction_files_count(files, COMPACTION_FILE_READ) != 0 || + compaction_files_count(files, COMPACTION_FILE_MODIFIED) != 0)))) || + epoch_record_preflight(epoch, files, retained) == -1 || + !retained_projection_valid(retained)) + return (-1); + system_epoch_init(&checked); + if (system_epoch_copy(&checked, epoch) == -1) + goto done; + buf_init(&b); + json_out_init(&jo, &b); + json_obj_begin(&jo); + json_key(&jo, "t"); + json_cstr(&jo, "system-epoch"); + json_key(&jo, "version"); + json_num(&jo, checked.version); + json_key(&jo, "reason"); + json_cstr(&jo, reason); + json_key(&jo, "encoding"); + json_cstr(&jo, "hex"); + json_key(&jo, "base"); + json_hex(&jo, &checked.base); + json_key(&jo, "personal"); + json_hex(&jo, &checked.personal); + json_key(&jo, "project"); + json_hex(&jo, &checked.project); + json_key(&jo, "summary"); + json_hex(&jo, &checked.summary); + json_key(&jo, "full"); + json_hex(&jo, &checked.full); + json_key(&jo, "restricted"); + json_hex(&jo, &checked.restricted); + json_key(&jo, "read_files"); + if (emit_file_array(&jo, files, COMPACTION_FILE_READ) == -1) + goto record_done; + json_key(&jo, "modified_files"); + if (emit_file_array(&jo, files, COMPACTION_FILE_MODIFIED) == -1) + goto record_done; + json_key(&jo, "retained"); + if (emit_retained(&jo, retained, &retained_count) == -1 || + (retained_count > 0 && checked.reason != SYSTEM_EPOCH_COMPACT)) + goto record_done; + if (turn_n > 0) { + json_key(&jo, "turn"); + json_num(&jo, turn_n); + } + json_obj_end(&jo); + if (b.len > JOURNAL_RECORD_MAX || + json_parse(&parsed, b.data, b.len, 0) != 0) + goto record_done; + json_done(&parsed); + append(j, &b); + ok = 0; + +record_done: + buf_free(&b); +done: + system_epoch_free(&checked); + return (ok); +} + void journal_turn_begin(struct journal *j, int n) { struct buf b; struct json_out jo; + if (j == NULL || n <= 0 || j->turn_n != 0) + fatalx("invalid journal Turn begin"); buf_init(&b); json_out_init(&jo, &b); json_obj_begin(&jo); @@ -294,6 +899,7 @@ journal_turn_begin(struct journal *j, int n) json_obj_end(&jo); append(j, &b); buf_free(&b); + j->turn_n = n; } void @@ -421,6 +1027,10 @@ journal_turn_end(struct journal *j, int n, enum turn_o struct json_out jo; const char *s; + if (j == NULL || n <= 0 || j->turn_n != n || + (outcome != TURN_OK && outcome != TURN_CAP && + outcome != TURN_ABANDONED)) + fatalx("invalid journal Turn end"); s = outcome == TURN_OK ? "ok" : outcome == TURN_CAP ? "cap" : "abandoned"; @@ -438,23 +1048,11 @@ journal_turn_end(struct journal *j, int n, enum turn_o buf_free(&b); /* durability at the transaction boundary (I14) */ journal_sync(j); + j->turn_n = 0; } /* --- replay ---------------------------------------------------- */ -static int -compact_record_valid(const struct json *jd, int root) -{ - char *s; - size_t len; - - s = json_get_str(jd, json_obj_get(jd, root, "summary"), &len); - if (s == NULL) - return (0); - free(s); - return (1); -} - /* Parse one "msg" record into a struct msg appended to buffered. */ static struct msg * parse_message(const struct json *jd, int obj) @@ -589,45 +1187,547 @@ parse_message(const struct json *jd, int obj) return (m); } -int -journal_replay(const char *dir, const char *id, struct msglist *out) +struct parsed_epoch { + struct system_epoch epoch; + struct compaction_files *files; + struct msglist retained; + int turn_n; +}; + +static void +parsed_epoch_init(struct parsed_epoch *parsed) { - struct msglist pending; /* current turn, not yet committed */ - struct msglist replayed; /* held until the read completes */ - struct msg *m, *tmp; + memset(parsed, 0, sizeof(*parsed)); + system_epoch_init(&parsed->epoch); + parsed->files = compaction_files_new(); + TAILQ_INIT(&parsed->retained); +} + +static void +parsed_epoch_free(struct parsed_epoch *parsed) +{ + system_epoch_free(&parsed->epoch); + compaction_files_free(parsed->files); + msglist_free(&parsed->retained); + memset(parsed, 0, sizeof(*parsed)); +} + +static int +token_string_is(const struct json *jd, int tok, const char *want) +{ + char *s; + size_t len, wantlen; + int equal; + + if ((s = json_get_str(jd, tok, &len)) == NULL) + return (0); + wantlen = strlen(want); + equal = len == wantlen && memcmp(s, want, len) == 0; + free(s); + return (equal); +} + +static int +hex_value(unsigned char c) +{ + if (c >= '0' && c <= '9') + return (c - '0'); + if (c >= 'a' && c <= 'f') + return (c - 'a' + 10); + if (c >= 'A' && c <= 'F') + return (c - 'A' + 10); + return (-1); +} + +static int +decode_hex(const struct json *jd, int tok, struct buf *out, size_t maxlen) +{ + char *s; + size_t i, len; + int hi, lo; + + if ((s = json_get_str(jd, tok, &len)) == NULL) + return (-1); + if ((len & 1) != 0 || len / 2 > maxlen) { + free(s); + return (-1); + } + buf_reset(out); + for (i = 0; i < len; i += 2) { + hi = hex_value((unsigned char)s[i]); + lo = hex_value((unsigned char)s[i + 1]); + if (hi == -1 || lo == -1) { + buf_reset(out); + free(s); + return (-1); + } + buf_addc(out, (hi << 4) | lo); + } + free(s); + return (0); +} + +static int +byte_cmp(const void *ap, size_t alen, const void *bp, size_t blen) +{ + size_t len; + int cmp; + + len = alen < blen ? alen : blen; + cmp = len == 0 ? 0 : memcmp(ap, bp, len); + if (cmp != 0) + return (cmp); + if (alen < blen) + return (-1); + if (alen > blen) + return (1); + return (0); +} + +static int +buf_same(const struct buf *a, const struct buf *b) +{ + return (a->len == b->len && + (a->len == 0 || memcmp(a->data, b->data, a->len) == 0)); +} + +static int +parse_file_array(const struct json *jd, int tok, + enum compaction_file_kind kind, struct compaction_files *files) +{ + struct buf previous; + char *path; + size_t before, i, len; + int item, count; + + if ((count = json_arr_len(jd, tok)) < 0 || + count > COMPACTION_FILE_COUNT_MAX) + return (-1); + buf_init(&previous); + for (i = 0; i < (size_t)count; i++) { + item = json_arr_get(jd, tok, (int)i); + if ((path = json_get_str(jd, item, &len)) == NULL) + goto fail; + if ((i > 0 && byte_cmp(previous.data, previous.len, path, len) + >= 0) || memchr(path, '\0', len) != NULL) { + free(path); + goto fail; + } + before = compaction_files_count(files, COMPACTION_FILE_READ) + + compaction_files_count(files, COMPACTION_FILE_MODIFIED); + if (compaction_files_add(files, kind, path, len) == -1 || + compaction_files_count(files, COMPACTION_FILE_READ) + + compaction_files_count(files, COMPACTION_FILE_MODIFIED) != + before + 1) { + free(path); + goto fail; + } + buf_reset(&previous); + buf_add(&previous, path, len); + free(path); + } + buf_free(&previous); + return (0); + +fail: + buf_free(&previous); + return (-1); +} + +static int +retained_block_key(const char *key, size_t len) +{ + static const char *const names[] = { + "type", "text", "id", "name", "input", "tool_use_id", + "content", "is_error" + }; + size_t i; + + for (i = 0; i < sizeof(names) / sizeof(names[0]); i++) + if (len == strlen(names[i]) && memcmp(key, names[i], len) == 0) + return ((int)i); + return (-1); +} + +static int +parse_retained_block(const struct json *jd, int obj, struct msg *m) +{ + struct buf a, b, c; + char *key; + size_t keylen; + unsigned int seen = 0; + int fields[8], i, id, n, value, is_error, ok = -1; + + for (i = 0; i < 8; i++) + fields[i] = -1; + if ((n = json_obj_len(jd, obj)) < 0) + return (-1); + for (i = 0; i < n; i++) { + if (json_obj_item(jd, obj, i, &key, &keylen, &value) == -1) + return (-1); + id = retained_block_key(key, keylen); + free(key); + if (id == -1 || (seen & (1U << id)) != 0) + return (-1); + seen |= 1U << id; + fields[id] = value; + } + if ((seen & 1U) == 0) + return (-1); + buf_init(&a); + buf_init(&b); + buf_init(&c); + if (token_string_is(jd, fields[0], "text")) { + if (seen != ((1U << 0) | (1U << 1)) || + decode_hex(jd, fields[1], &a, JOURNAL_RECORD_MAX / 2) == -1) + goto done; + msg_add_text(m, (const char *)a.data, a.len); + } else if (token_string_is(jd, fields[0], "tool-use")) { + if (m->role != ROLE_ASSISTANT || + seen != ((1U << 0) | (1U << 2) | (1U << 3) | (1U << 4)) || + decode_hex(jd, fields[2], &a, JOURNAL_RECORD_MAX / 2) == -1 || + decode_hex(jd, fields[3], &b, JOURNAL_RECORD_MAX / 2) == -1 || + decode_hex(jd, fields[4], &c, JOURNAL_RECORD_MAX / 2) == -1 || + a.len == 0 || b.len == 0 || + memchr(a.data, '\0', a.len) != NULL || + memchr(b.data, '\0', b.len) != NULL || + !json_valid_object(c.data, c.len)) + goto done; + msg_add_tool_use(m, buf_cstr(&a), buf_cstr(&b), + (const char *)c.data, c.len); + } else if (token_string_is(jd, fields[0], "tool-result")) { + if (m->role != ROLE_USER || + seen != ((1U << 0) | (1U << 5) | (1U << 6) | (1U << 7)) || + decode_hex(jd, fields[5], &a, JOURNAL_RECORD_MAX / 2) == -1 || + decode_hex(jd, fields[6], &b, JOURNAL_RECORD_MAX / 2) == -1 || + a.len == 0 || memchr(a.data, '\0', a.len) != NULL || + (is_error = json_get_bool(jd, fields[7])) == -1) + goto done; + msg_add_tool_result(m, buf_cstr(&a), (const char *)b.data, + b.len, is_error); + } else + goto done; + ok = 0; + +done: + buf_free(&a); + buf_free(&b); + buf_free(&c); + return (ok); +} + +static int +parse_retained_message(const struct json *jd, int obj, struct msg **out) +{ + struct msg *m = NULL; + char *key; + size_t keylen; + unsigned int seen = 0; + int role = -1, blocks = -1, block, i, n, value; + + *out = NULL; + if ((n = json_obj_len(jd, obj)) < 0) + return (-1); + for (i = 0; i < n; i++) { + if (json_obj_item(jd, obj, i, &key, &keylen, &value) == -1) + goto fail; + if (keylen == 4 && memcmp(key, "role", 4) == 0) { + if ((seen & 1U) != 0) { + free(key); + goto fail; + } + seen |= 1U; + role = value; + } else if (keylen == 6 && memcmp(key, "blocks", 6) == 0) { + if ((seen & 2U) != 0) { + free(key); + goto fail; + } + seen |= 2U; + blocks = value; + } else { + free(key); + goto fail; + } + free(key); + } + if (seen != 3U) + goto fail; + if (token_string_is(jd, role, "user")) + m = msg_new(ROLE_USER); + else if (token_string_is(jd, role, "assistant")) + m = msg_new(ROLE_ASSISTANT); + else + goto fail; + if ((n = json_arr_len(jd, blocks)) <= 0) + goto fail; + for (i = 0; i < n; i++) { + block = json_arr_get(jd, blocks, i); + if (parse_retained_block(jd, block, m) == -1) + goto fail; + } + *out = m; + return (0); + +fail: + if (m != NULL) + msg_free(m); + return (-1); +} + +static int +parse_retained_array(const struct json *jd, int tok, struct msglist *out) +{ + struct msg *m; + int i, n; + + if ((n = json_arr_len(jd, tok)) < 0) + return (-1); + for (i = 0; i < n; i++) { + if (parse_retained_message(jd, json_arr_get(jd, tok, i), &m) + == -1) + return (-1); + TAILQ_INSERT_TAIL(out, m, entry); + } + return (0); +} + +static int +epoch_key(const char *key, size_t len) +{ + static const char *const names[] = { + "t", "version", "reason", "encoding", "base", "personal", + "project", "summary", "full", "restricted", "read_files", + "modified_files", "retained", "turn" + }; + size_t i; + + for (i = 0; i < sizeof(names) / sizeof(names[0]); i++) + if (len == strlen(names[i]) && memcmp(key, names[i], len) == 0) + return ((int)i); + return (-1); +} + +static int +parse_epoch_record(const struct json *jd, int root, struct parsed_epoch *out) +{ + struct system_epoch candidate; + struct system_epoch_input input; + struct buf base, personal, project, summary, full, restricted; + char *key; + size_t keylen; + unsigned int seen = 0; + unsigned int required = ((1U << 10) - 1) | (1U << 12); + int64_t number; + int field[14], i, id, n, value, ok = -1; + + for (i = 0; i < 14; i++) + field[i] = -1; + if ((n = json_obj_len(jd, root)) < 0) + return (-1); + for (i = 0; i < n; i++) { + if (json_obj_item(jd, root, i, &key, &keylen, &value) == -1) + return (-1); + id = epoch_key(key, keylen); + free(key); + if (id == -1 || (seen & (1U << id)) != 0) + return (-1); + seen |= 1U << id; + field[id] = value; + } + if ((seen & required) != required || + !token_string_is(jd, field[0], "system-epoch") || + json_get_num(jd, field[1], &number) == -1 || + number != SYSTEM_EPOCH_VERSION || + !token_string_is(jd, field[3], "hex")) + return (-1); + if (token_string_is(jd, field[2], "start")) + out->epoch.reason = SYSTEM_EPOCH_START; + else if (token_string_is(jd, field[2], "clear")) + out->epoch.reason = SYSTEM_EPOCH_CLEAR; + else if (token_string_is(jd, field[2], "compact")) + out->epoch.reason = SYSTEM_EPOCH_COMPACT; + else if (token_string_is(jd, field[2], "migration")) + out->epoch.reason = SYSTEM_EPOCH_MIGRATION; + else + return (-1); + + buf_init(&base); + buf_init(&personal); + buf_init(&project); + buf_init(&summary); + buf_init(&full); + buf_init(&restricted); + system_epoch_init(&candidate); + if (decode_hex(jd, field[4], &base, SYSTEM_EPOCH_BASE_MAX) == -1 || + decode_hex(jd, field[5], &personal, + SYSTEM_EPOCH_EXTERNAL_MAX) == -1 || + decode_hex(jd, field[6], &project, + SYSTEM_EPOCH_EXTERNAL_MAX) == -1 || + decode_hex(jd, field[7], &summary, + SYSTEM_EPOCH_SUMMARY_MAX) == -1 || + decode_hex(jd, field[8], &full, + SYSTEM_EPOCH_VARIANT_MAX) == -1 || + decode_hex(jd, field[9], &restricted, + SYSTEM_EPOCH_VARIANT_MAX) == -1 || + (base.len > 0 && memchr(base.data, '\0', base.len) != NULL) || + (personal.len > 0 && memchr(personal.data, '\0', personal.len) + != NULL) || + (project.len > 0 && memchr(project.data, '\0', project.len) != NULL)) + goto done; + memset(&input, 0, sizeof(input)); + input.reason = out->epoch.reason; + input.base.data = base.data; + input.base.len = base.len; + input.personal.data = personal.data; + input.personal.len = personal.len; + input.project.data = project.data; + input.project.len = project.len; + input.summary.data = summary.data; + input.summary.len = summary.len; + if (system_epoch_build(&candidate, &input) == -1 || + !buf_same(&candidate.full, &full) || + !buf_same(&candidate.restricted, &restricted)) + goto done; + system_epoch_free(&out->epoch); + out->epoch = candidate; + system_epoch_init(&candidate); + if ((seen & (1U << 10)) != 0 && parse_file_array(jd, field[10], + COMPACTION_FILE_READ, out->files) == -1) + goto done; + if ((seen & (1U << 11)) != 0 && parse_file_array(jd, field[11], + COMPACTION_FILE_MODIFIED, out->files) == -1) + goto done; + if ((seen & (1U << 12)) != 0 && + parse_retained_array(jd, field[12], &out->retained) == -1) + goto done; + if (!retained_projection_valid(&out->retained) || + (out->epoch.reason != SYSTEM_EPOCH_COMPACT && + !TAILQ_EMPTY(&out->retained)) || + (out->epoch.reason == SYSTEM_EPOCH_CLEAR && + (out->epoch.summary.len != 0 || + compaction_files_count(out->files, COMPACTION_FILE_READ) != 0 || + compaction_files_count(out->files, COMPACTION_FILE_MODIFIED) != 0))) + goto done; + if ((seen & (1U << 13)) != 0) { + if (json_get_num(jd, field[13], &number) == -1 || number <= 0 || + number > INT_MAX || + out->epoch.reason != SYSTEM_EPOCH_COMPACT) + goto done; + out->turn_n = (int)number; + } + ok = 0; + +done: + buf_free(&base); + buf_free(&personal); + buf_free(&project); + buf_free(&summary); + buf_free(&full); + buf_free(&restricted); + system_epoch_free(&candidate); + return (ok); +} + +static void +messages_move(struct msglist *dst, struct msglist *src) +{ + struct msg *m; + + while ((m = TAILQ_FIRST(src)) != NULL) { + TAILQ_REMOVE(src, m, entry); + TAILQ_INSERT_TAIL(dst, m, entry); + } +} + +static void +parsed_epoch_move(struct parsed_epoch *dst, struct parsed_epoch *src) +{ + parsed_epoch_free(dst); + parsed_epoch_init(dst); + system_epoch_free(&dst->epoch); + dst->epoch = src->epoch; + memset(&src->epoch, 0, sizeof(src->epoch)); + compaction_files_free(dst->files); + dst->files = src->files; + src->files = NULL; + messages_move(&dst->retained, &src->retained); + dst->turn_n = src->turn_n; + src->turn_n = 0; +} + +static void +state_install_epoch(struct journal_state *state, struct parsed_epoch *parsed) +{ + system_epoch_free(&state->epoch); + state->epoch = parsed->epoch; + memset(&parsed->epoch, 0, sizeof(parsed->epoch)); + compaction_files_free(state->files); + state->files = parsed->files; + parsed->files = NULL; + state->have_epoch = 1; + buf_reset(&state->legacy_personal); + buf_reset(&state->legacy_project); + buf_reset(&state->legacy_summary); +} + +static void +pending_reset(struct msglist *pending, struct parsed_epoch *provisional, + int *have_provisional) +{ + msglist_free(pending); + TAILQ_INIT(pending); + parsed_epoch_free(provisional); + parsed_epoch_init(provisional); + *have_provisional = 0; +} + +static int +journal_scan_fd(int fd, struct journal_state *state) +{ + struct msglist pending; + struct parsed_epoch provisional; struct record_reader rr; struct buf line; - char path[1024]; int64_t pending_n = 0; - int fd, in_turn = 0, next, pending_bad = 0; + int have_provisional = 0, in_turn = 0, next; + int pending_bad = 0, saw_legacy = 0, saw_turn = 0; - if (!valid_id(id)) - return (-1); - if ((size_t)snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id) - >= sizeof(path)) - return (-1); - if ((fd = open(path, O_RDONLY)) == -1) - return (-1); - record_reader_init(&rr, fd); - buf_init(&line); TAILQ_INIT(&pending); - TAILQ_INIT(&replayed); - + parsed_epoch_init(&provisional); + record_reader_init(&rr, fd); + buf_init(&line); while ((next = record_next(&rr, &line)) != 0) { + struct parsed_epoch parsed; struct json jd; - int to; + struct msg *m; char *type; size_t tlen; + int root, to; if (next == RECORD_ERROR) - goto read_error; - if (next == RECORD_OVERSIZE || line.len == 0 || - json_parse(&jd, line.data, line.len, 0) != 0) { + goto fail; + if (next == RECORD_STRUCTURED_OVERSIZE) + goto fail; + if (next == RECORD_SEALED_TAIL) { if (in_turn) pending_bad = 1; - continue; /* tolerate blank/garbage lines */ + continue; } - to = json_obj_get(&jd, json_root(&jd), "t"); + if (next == RECORD_OVERSIZE || line.len == 0) { + if (in_turn) + pending_bad = 1; + continue; + } + if (json_parse(&jd, line.data, line.len, 0) != 0) { + if (line_mentions_epoch(line.data, line.len)) + goto fail; + if (in_turn) + pending_bad = 1; + continue; + } + root = json_root(&jd); + to = json_obj_get(&jd, root, "t"); type = to == -1 ? NULL : json_get_str(&jd, to, &tlen); if (type == NULL || strlen(type) != tlen) { if (in_turn) @@ -637,33 +1737,155 @@ journal_replay(const char *dir, const char *id, struct continue; } - if (strcmp(type, "clear") == 0 || - (strcmp(type, "compact") == 0 && - compact_record_valid(&jd, json_root(&jd)))) { - /* /clear and /compact both replace the active - * conversation. They also supersede a stale pending - * transaction left by a crash before a resumed process - * emitted the reset record. The compaction summary is - * re-injected separately (behavior.md 3). */ - msglist_free(&replayed); - TAILQ_INIT(&replayed); - msglist_free(&pending); - TAILQ_INIT(&pending); - in_turn = 0; - pending_bad = 0; - } else if (strcmp(type, "compact") == 0) { - if (in_turn) - pending_bad = 1; - } else if (strcmp(type, "turn-begin") == 0) { - int64_t n; - - if (json_get_num(&jd, json_obj_get(&jd, - json_root(&jd), "n"), &n) == 0 && - n > 0 && n <= INT_MAX) { - /* A new valid begin supersedes an uncommitted - * transaction left by a crashed process. */ + if (strcmp(type, "system-epoch") == 0) { + parsed_epoch_init(&parsed); + if (parse_epoch_record(&jd, root, &parsed) == -1) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + if (parsed.turn_n > 0) { + if (!state->have_epoch || !in_turn || + parsed.turn_n != pending_n) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } msglist_free(&pending); TAILQ_INIT(&pending); + parsed_epoch_move(&provisional, &parsed); + have_provisional = 1; + } else if (parsed.epoch.reason == SYSTEM_EPOCH_START) { + if (state->have_epoch || in_turn || saw_legacy || + saw_turn) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + state_install_epoch(state, &parsed); + parsed_epoch_free(&parsed); + } else if (parsed.epoch.reason == + SYSTEM_EPOCH_MIGRATION) { + if (state->have_epoch) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + pending_reset(&pending, &provisional, + &have_provisional); + in_turn = 0; + pending_bad = 0; + pending_n = 0; + state_install_epoch(state, &parsed); + parsed_epoch_free(&parsed); + } else { + if (!state->have_epoch) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + pending_reset(&pending, &provisional, + &have_provisional); + in_turn = 0; + pending_bad = 0; + pending_n = 0; + msglist_free(&state->messages); + TAILQ_INIT(&state->messages); + messages_move(&state->messages, &parsed.retained); + state_install_epoch(state, &parsed); + parsed_epoch_free(&parsed); + } + free(type); + json_done(&jd); + continue; + } + + if (strcmp(type, "context") == 0 || + strcmp(type, "clear") == 0 || + strcmp(type, "compact") == 0) { + if (state->have_epoch) { + free(type); + json_done(&jd); + goto fail; + } + if (strcmp(type, "context") == 0) { + char *personal, *project; + size_t plen, projectlen; + + personal = json_get_str(&jd, + json_obj_get(&jd, root, "personal"), &plen); + project = json_get_str(&jd, + json_obj_get(&jd, root, "project"), + &projectlen); + if (personal != NULL && project != NULL) { + buf_reset(&state->legacy_personal); + buf_reset(&state->legacy_project); + buf_add(&state->legacy_personal, personal, + plen); + buf_add(&state->legacy_project, project, + projectlen); + saw_legacy = 1; + } else if (in_turn) + pending_bad = 1; + free(personal); + free(project); + if (in_turn) + pending_bad = 1; + } else if (strcmp(type, "clear") == 0) { + saw_legacy = 1; + msglist_free(&state->messages); + TAILQ_INIT(&state->messages); + pending_reset(&pending, &provisional, + &have_provisional); + buf_reset(&state->legacy_personal); + buf_reset(&state->legacy_project); + buf_reset(&state->legacy_summary); + in_turn = 0; + pending_bad = 0; + pending_n = 0; + } else { + char *summary; + size_t len; + + summary = json_get_str(&jd, + json_obj_get(&jd, root, "summary"), &len); + if (summary != NULL) { + saw_legacy = 1; + msglist_free(&state->messages); + TAILQ_INIT(&state->messages); + pending_reset(&pending, &provisional, + &have_provisional); + buf_reset(&state->legacy_summary); + buf_add(&state->legacy_summary, summary, + len); + in_turn = 0; + pending_bad = 0; + pending_n = 0; + } else if (in_turn) + pending_bad = 1; + free(summary); + } + free(type); + json_done(&jd); + continue; + } + + if (strcmp(type, "turn-begin") == 0 || + strcmp(type, "msg") == 0 || + strcmp(type, "turn-end") == 0) + saw_turn = 1; + if (strcmp(type, "turn-begin") == 0) { + int64_t n; + + if (json_get_num(&jd, json_obj_get(&jd, root, "n"), + &n) == 0 && n > 0 && n <= INT_MAX) { + pending_reset(&pending, &provisional, + &have_provisional); in_turn = 1; pending_bad = 0; pending_n = n; @@ -671,8 +1893,7 @@ journal_replay(const char *dir, const char *id, struct pending_bad = 1; } else if (strcmp(type, "msg") == 0) { if (in_turn) { - if ((m = parse_message(&jd, - json_root(&jd))) != NULL) + if ((m = parse_message(&jd, root)) != NULL) TAILQ_INSERT_TAIL(&pending, m, entry); else pending_bad = 1; @@ -680,28 +1901,29 @@ journal_replay(const char *dir, const char *id, struct } else if (strcmp(type, "turn-end") == 0) { char *oc; size_t ol; - int64_t n; + int64_t n; int commit; oc = json_get_str(&jd, - json_obj_get(&jd, json_root(&jd), "outcome"), - &ol); + json_obj_get(&jd, root, "outcome"), &ol); commit = in_turn && !pending_bad && - json_get_num(&jd, json_obj_get(&jd, - json_root(&jd), "n"), &n) == 0 && - n == pending_n && oc != NULL && + json_get_num(&jd, json_obj_get(&jd, root, "n"), + &n) == 0 && n == pending_n && oc != NULL && ((ol == 2 && memcmp(oc, "ok", 2) == 0) || (ol == 3 && memcmp(oc, "cap", 3) == 0)); free(oc); if (commit) { - TAILQ_FOREACH_SAFE(m, &pending, entry, tmp) { - TAILQ_REMOVE(&pending, m, entry); - TAILQ_INSERT_TAIL(&replayed, m, entry); + if (have_provisional) { + msglist_free(&state->messages); + TAILQ_INIT(&state->messages); + messages_move(&state->messages, + &provisional.retained); + state_install_epoch(state, &provisional); } - } else { - msglist_free(&pending); - TAILQ_INIT(&pending); + messages_move(&state->messages, &pending); } + pending_reset(&pending, &provisional, + &have_provisional); in_turn = 0; pending_bad = 0; pending_n = 0; @@ -710,26 +1932,76 @@ journal_replay(const char *dir, const char *id, struct free(type); json_done(&jd); } - /* a trailing partial (unterminated) line is ignored */ msglist_free(&pending); - TAILQ_FOREACH_SAFE(m, &replayed, entry, tmp) { - TAILQ_REMOVE(&replayed, m, entry); - TAILQ_INSERT_TAIL(out, m, entry); - } + parsed_epoch_free(&provisional); buf_free(&line); record_reader_done(&rr); - close(fd); return (0); -read_error: +fail: msglist_free(&pending); - msglist_free(&replayed); + parsed_epoch_free(&provisional); buf_free(&line); record_reader_done(&rr); - close(fd); return (-1); } +int +journal_restore(const char *dir, const char *id, struct journal_state *state) +{ + struct journal_state next; + char path[1024]; + int fd, ok; + + if (!valid_id(id) || (size_t)snprintf(path, sizeof(path), + "%s/%s.ndjson", dir, id) >= sizeof(path) || + (fd = open(path, O_RDONLY)) == -1) + return (-1); + journal_state_init(&next); + ok = journal_scan_fd(fd, &next); + close(fd); + if (ok == -1) { + journal_state_free(&next); + return (-1); + } + journal_state_free(state); + journal_state_init(state); + messages_move(&state->messages, &next.messages); + system_epoch_free(&state->epoch); + state->epoch = next.epoch; + memset(&next.epoch, 0, sizeof(next.epoch)); + compaction_files_free(state->files); + state->files = next.files; + next.files = NULL; + buf_free(&state->legacy_personal); + state->legacy_personal = next.legacy_personal; + buf_init(&next.legacy_personal); + buf_free(&state->legacy_project); + state->legacy_project = next.legacy_project; + buf_init(&next.legacy_project); + buf_free(&state->legacy_summary); + state->legacy_summary = next.legacy_summary; + buf_init(&next.legacy_summary); + state->have_epoch = next.have_epoch; + journal_state_free(&next); + return (0); +} + +int +journal_replay(const char *dir, const char *id, struct msglist *out) +{ + struct journal_state state; + + journal_state_init(&state); + if (journal_restore(dir, id, &state) == -1) { + journal_state_free(&state); + return (-1); + } + messages_move(out, &state.messages); + journal_state_free(&state); + return (0); +} + /* * Fill personal/project (reset then appended) with the bytes of the * LAST context event in the session, so a resumed session injects the @@ -740,92 +2012,39 @@ int journal_replay_context(const char *dir, const char *id, struct buf *personal, struct buf *project, struct buf *compaction) { - struct record_reader rr; - struct buf line; - char path[1024]; - int fd, next; + struct journal_state state; buf_reset(personal); buf_reset(project); buf_reset(compaction); - if (!valid_id(id)) + journal_state_init(&state); + if (journal_restore(dir, id, &state) == -1) { + journal_state_free(&state); return (-1); - if ((size_t)snprintf(path, sizeof(path), "%s/%s.ndjson", dir, id) >= - sizeof(path)) - return (-1); - if ((fd = open(path, O_RDONLY)) == -1) - return (-1); - record_reader_init(&rr, fd); - buf_init(&line); - while ((next = record_next(&rr, &line)) != 0) { - struct json jd; - int root, to; - char *type; - size_t tl; - - if (next == RECORD_ERROR) - goto read_error; - if (next == RECORD_OVERSIZE || line.len == 0 || - json_parse(&jd, line.data, line.len, 0) != 0) - continue; - root = json_root(&jd); - to = json_obj_get(&jd, root, "t"); - type = to == -1 ? NULL : json_get_str(&jd, to, &tl); - if (type != NULL && strlen(type) != tl) { - free(type); - json_done(&jd); - continue; - } - if (type != NULL && strcmp(type, "context") == 0) { - char *pe, *pr; - size_t pel, prl; - - pe = json_get_str(&jd, - json_obj_get(&jd, root, "personal"), &pel); - pr = json_get_str(&jd, - json_obj_get(&jd, root, "project"), &prl); - if (pe != NULL && pr != NULL) { - buf_reset(personal); - buf_reset(project); - buf_add(personal, pe, pel); - buf_add(project, pr, prl); - } - free(pe); - free(pr); - } else if (type != NULL && strcmp(type, "compact") == 0) { - char *s; - size_t sl; - - s = json_get_str(&jd, - json_obj_get(&jd, root, "summary"), &sl); - if (s != NULL) { - buf_reset(compaction); - buf_add(compaction, s, sl); - } - free(s); - } else if (type != NULL && strcmp(type, "clear") == 0) { - /* If a crash lands between the durable reset and its new - * context snapshot, never resurrect pre-clear context. */ - buf_reset(personal); - buf_reset(project); - buf_reset(compaction); - } - free(type); - json_done(&jd); } - buf_free(&line); - record_reader_done(&rr); - close(fd); + if (state.have_epoch) { + if (state.epoch.personal.len > 0) + buf_add(personal, state.epoch.personal.data, + state.epoch.personal.len); + if (state.epoch.project.len > 0) + buf_add(project, state.epoch.project.data, + state.epoch.project.len); + if (state.epoch.summary.len > 0) + buf_add(compaction, state.epoch.summary.data, + state.epoch.summary.len); + } else { + if (state.legacy_personal.len > 0) + buf_add(personal, state.legacy_personal.data, + state.legacy_personal.len); + if (state.legacy_project.len > 0) + buf_add(project, state.legacy_project.data, + state.legacy_project.len); + if (state.legacy_summary.len > 0) + buf_add(compaction, state.legacy_summary.data, + state.legacy_summary.len); + } + journal_state_free(&state); return (0); - -read_error: - buf_reset(personal); - buf_reset(project); - buf_reset(compaction); - buf_free(&line); - record_reader_done(&rr); - close(fd); - return (-1); } int @@ -921,49 +2140,138 @@ preview_text(const char *text, size_t len, char *out, out[o] = '\0'; } +struct projection_summary { + int64_t messages; + int have_preview; + char preview[sizeof(((struct session_ent *)0)->preview)]; +}; + +static void +projection_summary_reset(struct projection_summary *summary) +{ + memset(summary, 0, sizeof(*summary)); +} + +static int +projection_summary_add(struct projection_summary *summary, + const struct msg *message) +{ + const struct block *block; + + if (summary->messages == INT64_MAX) + return (-1); + summary->messages++; + if (summary->have_preview || message->role != ROLE_USER) + return (0); + TAILQ_FOREACH(block, &message->blocks, entry) { + if (block->type != BLOCK_TEXT) + continue; + preview_text(block->text, block->textlen, summary->preview, + sizeof(summary->preview)); + summary->have_preview = 1; + break; + } + return (0); +} + +static int +projection_summary_messages(struct projection_summary *summary, + const struct msglist *messages) +{ + const struct msg *message; + + projection_summary_reset(summary); + TAILQ_FOREACH(message, messages, entry) + if (projection_summary_add(summary, message) == -1) + return (-1); + return (0); +} + +static int +projection_summary_append(struct projection_summary *summary, + const struct projection_summary *suffix) +{ + if (suffix->messages > INT64_MAX - summary->messages) + return (-1); + summary->messages += suffix->messages; + if (!summary->have_preview && suffix->have_preview) { + strlcpy(summary->preview, suffix->preview, + sizeof(summary->preview)); + summary->have_preview = 1; + } + return (0); +} + +static void +projection_summary_publish(const struct projection_summary *summary, + struct session_ent *e) +{ + e->messages = summary->messages; + if (summary->have_preview) + strlcpy(e->preview, summary->preview, sizeof(e->preview)); + else + e->preview[0] = '\0'; +} + /* - * Summarize only committed turns. Journals are transactional (I14), so - * neither an abandoned turn nor a crash-mid-turn contributes to -l. - * The scan is record-bounded, not whole-file-bounded, and ignores a partial - * final line so long append-only sessions remain listable. + * Fold only committed summary state. At most the current NDJSON record is + * decoded: ordinary messages are counted and freed immediately, while one + * epoch's retained Projection is summarized before that parsed record is + * released. Neither the live Projection nor a complete Journal is retained. */ static int session_summary(const char *path, off_t size, struct session_ent *e) { + struct projection_summary current, pending, provisional; struct record_reader rr; struct buf line; - char pending_preview[sizeof(e->preview)]; - int64_t pending = 0, pending_n = 0; - int fd, next, in_turn = 0, pending_bad = 0; - int have_preview = 0, pending_have_preview = 0; + int64_t pending_n = 0; + int fd, have_epoch = 0, have_provisional = 0; + int in_turn = 0, next, pending_bad = 0; + int saw_legacy = 0, saw_turn = 0; e->messages = 0; e->preview[0] = '\0'; if (size < 0 || (fd = open(path, O_RDONLY)) == -1) return (-1); + projection_summary_reset(¤t); + projection_summary_reset(&pending); + projection_summary_reset(&provisional); record_reader_init(&rr, fd); buf_init(&line); - pending_preview[0] = '\0'; while ((next = record_next(&rr, &line)) != 0) { + struct parsed_epoch parsed; + struct projection_summary epoch_summary; struct json jd; - struct msg *m; - struct block *b; - char *type = NULL; - size_t tlen = 0; - int root; + struct msg *message; + char *type; + size_t tlen; + int root, to; - if (next == RECORD_ERROR) - goto read_error; - if (next == RECORD_OVERSIZE || line.len == 0 || - json_parse(&jd, line.data, line.len, 0) != 0) { + if (next == RECORD_ERROR || + next == RECORD_STRUCTURED_OVERSIZE) + goto fail; + if (next == RECORD_SEALED_TAIL) { if (in_turn) pending_bad = 1; continue; } + if (next == RECORD_OVERSIZE || line.len == 0) { + if (in_turn) + pending_bad = 1; + continue; + } + if (json_parse(&jd, line.data, line.len, 0) != 0) { + if (line_mentions_epoch(line.data, line.len)) + goto fail; + if (in_turn) + pending_bad = 1; + continue; + } root = json_root(&jd); - type = json_get_str(&jd, json_obj_get(&jd, root, "t"), - &tlen); + to = json_obj_get(&jd, root, "t"); + type = to == -1 ? NULL : json_get_str(&jd, to, &tlen); if (type == NULL || strlen(type) != tlen) { if (in_turn) pending_bad = 1; @@ -971,62 +2279,161 @@ session_summary(const char *path, off_t size, struct s json_done(&jd); continue; } - if (strcmp(type, "clear") == 0 || - (strcmp(type, "compact") == 0 && - compact_record_valid(&jd, root))) { - e->messages = 0; - e->preview[0] = '\0'; - have_preview = 0; - pending = 0; - pending_have_preview = 0; - pending_preview[0] = '\0'; - pending_n = 0; - in_turn = 0; - pending_bad = 0; - } else if (strcmp(type, "compact") == 0) { - if (in_turn) - pending_bad = 1; - } else if (strcmp(type, "turn-begin") == 0) { - int64_t n; + if (strcmp(type, "system-epoch") == 0) { + parsed_epoch_init(&parsed); + if (parse_epoch_record(&jd, root, &parsed) == -1 || + projection_summary_messages(&epoch_summary, + &parsed.retained) == -1) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + if (parsed.turn_n > 0) { + if (!have_epoch || !in_turn || + parsed.turn_n != pending_n) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + projection_summary_reset(&pending); + provisional = epoch_summary; + have_provisional = 1; + } else if (parsed.epoch.reason == SYSTEM_EPOCH_START) { + if (have_epoch || in_turn || saw_legacy || saw_turn) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + have_epoch = 1; + } else if (parsed.epoch.reason == + SYSTEM_EPOCH_MIGRATION) { + if (have_epoch) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + projection_summary_reset(&pending); + projection_summary_reset(&provisional); + have_provisional = 0; + in_turn = 0; + pending_bad = 0; + pending_n = 0; + have_epoch = 1; + } else { + if (!have_epoch) { + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + goto fail; + } + projection_summary_reset(&pending); + projection_summary_reset(&provisional); + have_provisional = 0; + in_turn = 0; + pending_bad = 0; + pending_n = 0; + current = epoch_summary; + } + parsed_epoch_free(&parsed); + free(type); + json_done(&jd); + continue; + } + + if (strcmp(type, "context") == 0 || + strcmp(type, "clear") == 0 || + strcmp(type, "compact") == 0) { + if (have_epoch) { + free(type); + json_done(&jd); + goto fail; + } + if (strcmp(type, "context") == 0) { + char *personal, *project; + size_t plen, projectlen; + + personal = json_get_str(&jd, + json_obj_get(&jd, root, "personal"), &plen); + project = json_get_str(&jd, + json_obj_get(&jd, root, "project"), + &projectlen); + if (personal != NULL && project != NULL) + saw_legacy = 1; + else if (in_turn) + pending_bad = 1; + free(personal); + free(project); + if (in_turn) + pending_bad = 1; + } else if (strcmp(type, "clear") == 0) { + saw_legacy = 1; + projection_summary_reset(¤t); + projection_summary_reset(&pending); + projection_summary_reset(&provisional); + have_provisional = 0; + in_turn = 0; + pending_bad = 0; + pending_n = 0; + } else { + char *summary; + size_t len; + + summary = json_get_str(&jd, + json_obj_get(&jd, root, "summary"), &len); + if (summary != NULL) { + saw_legacy = 1; + projection_summary_reset(¤t); + projection_summary_reset(&pending); + projection_summary_reset(&provisional); + have_provisional = 0; + in_turn = 0; + pending_bad = 0; + pending_n = 0; + } else if (in_turn) + pending_bad = 1; + free(summary); + } + free(type); + json_done(&jd); + continue; + } + + if (strcmp(type, "turn-begin") == 0 || + strcmp(type, "msg") == 0 || + strcmp(type, "turn-end") == 0) + saw_turn = 1; + if (strcmp(type, "turn-begin") == 0) { + int64_t n; + if (json_get_num(&jd, json_obj_get(&jd, root, "n"), &n) == 0 && n > 0 && n <= INT_MAX) { - pending = 0; - pending_have_preview = 0; - pending_preview[0] = '\0'; - pending_n = n; + projection_summary_reset(&pending); + projection_summary_reset(&provisional); + have_provisional = 0; in_turn = 1; pending_bad = 0; + pending_n = n; } else if (in_turn) pending_bad = 1; } else if (strcmp(type, "msg") == 0) { - if (!in_turn) - goto record_done; - m = parse_message(&jd, root); - if (m == NULL) { - pending_bad = 1; - goto record_done; - } - if (pending == INT64_MAX) { - pending_bad = 1; - msg_free(m); - goto record_done; - } - pending++; - if (!have_preview && !pending_have_preview && - m->role == ROLE_USER) { - TAILQ_FOREACH(b, &m->blocks, entry) { - if (b->type != BLOCK_TEXT) - continue; - preview_text(b->text, b->textlen, - pending_preview, - sizeof(pending_preview)); - pending_have_preview = 1; - break; + if (in_turn) { + message = parse_message(&jd, root); + if (message == NULL) + pending_bad = 1; + else { + if (projection_summary_add(&pending, + message) == -1) + pending_bad = 1; + msg_free(message); } } - msg_free(m); } else if (strcmp(type, "turn-end") == 0) { + struct projection_summary committed; char *oc; size_t ol; int64_t n; @@ -1038,35 +2445,36 @@ session_summary(const char *path, off_t size, struct s json_get_num(&jd, json_obj_get(&jd, root, "n"), &n) == 0 && n == pending_n && oc != NULL && ((ol == 2 && memcmp(oc, "ok", 2) == 0) || - (ol == 3 && memcmp(oc, "cap", 3) == 0)) && - pending <= INT64_MAX - e->messages; - if (commit) { - e->messages += pending; - if (!have_preview && pending_have_preview) { - strlcpy(e->preview, pending_preview, - sizeof(e->preview)); - have_preview = 1; - } - } + (ol == 3 && memcmp(oc, "cap", 3) == 0)); free(oc); - pending = 0; - pending_have_preview = 0; - pending_preview[0] = '\0'; - pending_n = 0; + if (commit) { + committed = have_provisional ? provisional : current; + if (projection_summary_append(&committed, + &pending) == -1) { + free(type); + json_done(&jd); + goto fail; + } + current = committed; + } + projection_summary_reset(&pending); + projection_summary_reset(&provisional); + have_provisional = 0; in_turn = 0; pending_bad = 0; + pending_n = 0; } else if (in_turn) pending_bad = 1; -record_done: free(type); json_done(&jd); } + projection_summary_publish(¤t, e); buf_free(&line); record_reader_done(&rr); close(fd); return (0); -read_error: +fail: e->messages = 0; e->preview[0] = '\0'; buf_free(&line); blob - 86a47a7972d8620984bf80405b25f404d332b391 blob + f779a4aa2e1e23572e90739b082ea0672b7d3a8e --- src/fugu/journal.h +++ src/fugu/journal.h @@ -21,6 +21,7 @@ #include "buf.h" #include "msg.h" +#include "system_epoch.h" /* * The session journal (behavior.md section 7, invariants.md I14). @@ -33,6 +34,7 @@ */ #define SESSION_ID_MAX 32 /* "YYYYMMDDThhmmss-xxxx" + slack */ +#define JOURNAL_RECORD_MAX (32 * 1024 * 1024) /* excluding newline */ enum turn_outcome { TURN_OK, /* completed normally */ @@ -41,7 +43,27 @@ enum turn_outcome { }; struct journal; +struct compaction_files; +/* + * One owned resume result. When have_epoch is true, epoch and files are + * the last durable System epoch and its tracked file sets. Otherwise the + * legacy buffers are the folded context/compact state of an old journal. + * messages is always the committed Projection after the last reset. + */ +struct journal_state { + struct msglist messages; + struct system_epoch epoch; + struct compaction_files *files; + struct buf legacy_personal; + struct buf legacy_project; + struct buf legacy_summary; + int have_epoch; +}; + +void journal_state_init(struct journal_state *); +void journal_state_free(struct journal_state *); + /* Resolve ~/.fugu/sessions, creating ~/.fugu and it (0700) as needed. */ int journal_dir(char *, size_t); @@ -58,6 +80,18 @@ void journal_turn_end(struct journal *, int n, enum tu /* Make all preceding records durable; failure is fatal. */ void journal_sync(struct journal *); +/* + * Append one complete, record-bounded System epoch snapshot. retained is + * the optional post-reset Projection snapshot. turn_n is zero for an + * immediate boundary, or the matching active Turn number for a provisional + * compact boundary committed by that Turn's ok/cap turn-end. + */ +int journal_system_epoch(struct journal *, const struct system_epoch *, + const struct compaction_files *, const struct msglist *, int); + +/* Atomically replace an initialized state with the restored owned state. */ +int journal_restore(const char *, const char *, struct journal_state *); + /* Record a context reset (/clear): replay after it starts fresh. */ void journal_clear(struct journal *); @@ -74,7 +108,7 @@ int journal_replay_context(const char *dir, const char * Replay a session into out (appended). Only committed turns are * reproduced; the projection is left in a state the provider accepts * (no orphaned tool calls, no adjacent same-role messages). Returns - * 0 on success, -1 if the session does not exist. + * 0 on success, -1 if the session is unavailable or invalid. */ int journal_replay(const char *dir, const char *id, struct msglist *out); blob - 78d9dfe99abc788cbd205f9c9dd5c32be8b083de blob + a316eee11edbae7a141c7cf833794ec9f1b273d8 --- src/fugu/turn_txn.c +++ src/fugu/turn_txn.c @@ -20,10 +20,13 @@ #include #include #include +#include +#include "compaction.h" #include "log.h" #include "msg.h" #include "journal.h" +#include "system_epoch.h" #include "xmalloc.h" #include "turn_txn.h" #include "turn_txn_private.h" @@ -37,9 +40,12 @@ struct turn_txn { struct msglist messages; size_t message_count; struct msg *pending_first; + struct msglist rollback; + size_t rollback_count; size_t sent_prefix_len; uint64_t snapshot_version; int turn_n; + int have_rollback; enum turn_txn_state state; struct journal *journal; struct turn_txn_jobs jobs; @@ -105,6 +111,7 @@ turn_txn_new(struct journal *journal, struct msglist * } txn = xcalloc(1, sizeof(*txn)); TAILQ_INIT(&txn->messages); + TAILQ_INIT(&txn->rollback); TAILQ_CONCAT(&txn->messages, seed, entry); txn->message_count = count; txn->snapshot_version = 1; @@ -120,6 +127,7 @@ turn_txn_free(struct turn_txn *txn) return; require_idle(txn); msglist_free(&txn->messages); + msglist_free(&txn->rollback); free(txn); } @@ -205,6 +213,10 @@ turn_txn_commit(struct turn_txn *txn, enum turn_txn_co journal_turn_end(txn->journal, txn->turn_n, journal_outcome); txn->jobs.commit(txn->jobs.arg); txn->pending_first = NULL; + msglist_free(&txn->rollback); + TAILQ_INIT(&txn->rollback); + txn->rollback_count = 0; + txn->have_rollback = 0; txn->state = TURN_TXN_IDLE; snapshot_changed(txn); } @@ -232,7 +244,16 @@ turn_txn_abandon(struct turn_txn *txn) txn->jobs.stop_executor_if_live(txn->jobs.arg); txn->jobs.abort(txn->jobs.arg); journal_turn_end(txn->journal, txn->turn_n, TURN_ABANDONED); - discard_pending(txn); + if (txn->have_rollback) { + msglist_free(&txn->messages); + TAILQ_INIT(&txn->messages); + TAILQ_CONCAT(&txn->messages, &txn->rollback, entry); + txn->message_count = txn->rollback_count; + txn->rollback_count = 0; + txn->have_rollback = 0; + txn->pending_first = NULL; + } else + discard_pending(txn); txn->sent_prefix_len = 0; txn->state = TURN_TXN_IDLE; snapshot_changed(txn); @@ -246,41 +267,154 @@ turn_txn_cache_reset(struct turn_txn *txn) snapshot_changed(txn); } -static void -reset_projection(struct turn_txn *txn) +static struct msg * +message_clone(const struct msg *source) { + const struct block *block; + struct msg *copy; + + if (source == NULL || (source->role != ROLE_USER && + source->role != ROLE_ASSISTANT)) + return (NULL); + copy = msg_new(source->role); + TAILQ_FOREACH(block, &source->blocks, entry) { + switch (block->type) { + case BLOCK_TEXT: + if (block->text == NULL && block->textlen != 0) { + msg_free(copy); + return (NULL); + } + msg_add_text(copy, block->text, block->textlen); + break; + case BLOCK_TOOL_USE: + if (block->tool_id == NULL || block->tool_name == NULL || + (block->tool_input == NULL && + block->tool_input_len != 0)) { + msg_free(copy); + return (NULL); + } + msg_add_tool_use(copy, block->tool_id, block->tool_name, + block->tool_input, block->tool_input_len); + break; + case BLOCK_TOOL_RESULT: + if (block->tool_use_id == NULL || + (block->result == NULL && block->result_len != 0) || + (block->is_error != 0 && block->is_error != 1)) { + msg_free(copy); + return (NULL); + } + msg_add_tool_result(copy, block->tool_use_id, + block->result, block->result_len, block->is_error); + break; + default: + msg_free(copy); + return (NULL); + } + } + return (copy); +} + +static int +messages_clone_range(const struct msglist *source, size_t first, size_t limit, + struct msglist *out, size_t *countp) +{ + const struct msg *message; + struct msg *copy; + size_t at = 0, count = 0; + + TAILQ_INIT(out); + TAILQ_FOREACH(message, source, entry) { + if (at++ < first) + continue; + if (count == limit) + break; + if ((copy = message_clone(message)) == NULL) { + msglist_free(out); + TAILQ_INIT(out); + return (-1); + } + TAILQ_INSERT_TAIL(out, copy, entry); + count++; + } + if (count != limit) { + msglist_free(out); + TAILQ_INIT(out); + return (-1); + } + *countp = count; + return (0); +} + +static int +preturn_count(const struct turn_txn *txn, size_t *countp) +{ + const struct msg *message; + size_t count = 0; + + if (txn->pending_first == NULL) { + *countp = txn->message_count; + return (0); + } + TAILQ_FOREACH(message, &txn->messages, entry) { + if (message == txn->pending_first) { + *countp = count; + return (0); + } + count++; + } + return (-1); +} + +int +turn_txn_epoch_reset(struct turn_txn *txn, const struct system_epoch *epoch, + const struct compaction_files *files, size_t retain_first) +{ + struct msglist retained, rollback; + size_t retained_count, rollback_count = 0; + int active, need_rollback; + + require_txn(txn); + active = txn->state == TURN_TXN_ACTIVE; + if (epoch == NULL || files == NULL || retain_first > txn->message_count || + (epoch->reason != SYSTEM_EPOCH_CLEAR && + epoch->reason != SYSTEM_EPOCH_COMPACT) || + (active && epoch->reason != SYSTEM_EPOCH_COMPACT) || + (epoch->reason == SYSTEM_EPOCH_CLEAR && + (retain_first != txn->message_count || epoch->summary.len != 0))) + return (-1); + retained_count = txn->message_count - retain_first; + if (messages_clone_range(&txn->messages, retain_first, retained_count, + &retained, &retained_count) == -1) + return (-1); + TAILQ_INIT(&rollback); + need_rollback = active && !txn->have_rollback; + if (need_rollback) { + if (preturn_count(txn, &rollback_count) == -1 || + messages_clone_range(&txn->messages, 0, rollback_count, + &rollback, &rollback_count) == -1) { + msglist_free(&retained); + return (-1); + } + } + if (journal_system_epoch(txn->journal, epoch, files, &retained, + active ? txn->turn_n : 0) == -1) { + msglist_free(&rollback); + msglist_free(&retained); + return (-1); + } + journal_sync(txn->journal); + + if (need_rollback) { + TAILQ_CONCAT(&txn->rollback, &rollback, entry); + txn->rollback_count = rollback_count; + txn->have_rollback = 1; + } msglist_free(&txn->messages); TAILQ_INIT(&txn->messages); - txn->message_count = 0; + TAILQ_CONCAT(&txn->messages, &retained, entry); + txn->message_count = retained_count; txn->pending_first = NULL; txn->sent_prefix_len = 0; snapshot_changed(txn); + return (0); } - -void -turn_txn_clear(struct turn_txn *txn, const struct turn_txn_context *context) -{ - require_idle(txn); - if (context != NULL && - ((context->personal == NULL && context->personal_len != 0) || - (context->project == NULL && context->project_len != 0))) - fatalx("invalid Turn transaction context span"); - journal_clear(txn->journal); - if (context != NULL) - journal_context(txn->journal, context->personal, - context->personal_len, context->project, - context->project_len); - journal_sync(txn->journal); - reset_projection(txn); -} - -void -turn_txn_compact(struct turn_txn *txn, const void *summary, size_t len) -{ - require_idle(txn); - if (summary == NULL && len != 0) - fatalx("invalid Turn transaction compaction span"); - journal_compact(txn->journal, summary, len); - journal_sync(txn->journal); - reset_projection(txn); -} blob - bb2d9bf523aa772d6df503f3c7dfa1ba0ae55fc7 blob + c2626b086691e9f081cd7b785ac33d41ed7d42c7 --- src/fugu/turn_txn.h +++ src/fugu/turn_txn.h @@ -23,6 +23,8 @@ struct msg; struct msglist; +struct compaction_files; +struct system_epoch; struct turn_txn; enum turn_txn_commit { @@ -42,13 +44,6 @@ struct turn_txn_projection { uint64_t token; }; -struct turn_txn_context { - const char *personal; - size_t personal_len; - const char *project; - size_t project_len; -}; - void turn_txn_free(struct turn_txn *); void turn_txn_begin(struct turn_txn *); /* Transfers message ownership exactly once. */ @@ -64,7 +59,16 @@ void turn_txn_abandon(struct turn_txn *); * Valid while idle or active; invalidates every outstanding receipt. */ void turn_txn_cache_reset(struct turn_txn *); -void turn_txn_clear(struct turn_txn *, const struct turn_txn_context *); -void turn_txn_compact(struct turn_txn *, const void *, size_t); +/* + * Install a clear or Compaction System epoch and discard the leading + * retain_first messages. The retained suffix is recorded before the live + * Projection changes. A Compaction may run inside an active Turn; its + * Journal epoch then remains provisional until that Turn commits. + * + * Returns -1, without changing the Projection or cursor, when the epoch, + * file state, cut, or transaction-state combination is invalid. + */ +int turn_txn_epoch_reset(struct turn_txn *, const struct system_epoch *, + const struct compaction_files *, size_t); #endif /* TURN_TXN_H */