commit - 3f83e196092e9ddcb3fcece455be1b57ab878757
commit + f3f30b82b058e173347141ecf1ccdebb3a6b2a16
blob - 998feba74304e5deefac478cc6ac46c6bf9904d1
blob + 5dfc92e6fd593cc61f756f4fca14223e95f8dd8e
--- regress/journal/Makefile
+++ regress/journal/Makefile
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
#include <sys/types.h>
#include <sys/queue.h>
+#include <sys/resource.h>
#include <sys/stat.h>
+#include <sys/wait.h>
#include <err.h>
+#include <errno.h>
#include <fcntl.h>
+#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#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)
{
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;
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[] =
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
}
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
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
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
#include <syslog.h>
#include "buf.h"
+#include "compaction.h"
#include "journal.h"
#include "log.h"
#include "msg.h"
}
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};
}
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
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
#include <syslog.h>
#include <unistd.h>
+#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"
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
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)
{
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)
}
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,
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,
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;
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);
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
#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));
/*
* 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)
(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;
}
}
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);
}
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);
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);
}
}
}
+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);
json_obj_end(&jo);
append(j, &b);
buf_free(&b);
+ j->turn_n = n;
}
void
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";
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)
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)
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;
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;
} 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;
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
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
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;
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;
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
#include "buf.h"
#include "msg.h"
+#include "system_epoch.h"
/*
* The session journal (behavior.md section 7, invariants.md I14).
*/
#define SESSION_ID_MAX 32 /* "YYYYMMDDThhmmss-xxxx" + slack */
+#define JOURNAL_RECORD_MAX (32 * 1024 * 1024) /* excluding newline */
enum turn_outcome {
TURN_OK, /* completed normally */
};
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);
/* 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 *);
* 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
#include <limits.h>
#include <stdint.h>
#include <stdlib.h>
+#include <string.h>
+#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"
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;
}
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;
return;
require_idle(txn);
msglist_free(&txn->messages);
+ msglist_free(&txn->rollback);
free(txn);
}
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);
}
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);
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
struct msg;
struct msglist;
+struct compaction_files;
+struct system_epoch;
struct turn_txn;
enum turn_txn_commit {
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. */
* 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 */
