niklas/amduat
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: niklas:8ea3ec861fa56dc2d30e544ae9487a03bc93c272
Branches Tags
niklas:main
niklas:niklas/lock-file-fix-anchor
niklas:apiv2
niklas:federation
niklas:qbits
niklas:burgen-fels
..
compare: niklas:26fa4cce667b1833475a7aead68478fd230b9172
Branches Tags
niklas:main
niklas:niklas/lock-file-fix-anchor
niklas:apiv2
niklas:federation
niklas:qbits
niklas:burgen-fels

No commits in common. "8ea3ec861fa56dc2d30e544ae9487a03bc93c272" and "26fa4cce667b1833475a7aead68478fd230b9172" have entirely different histories.
8ea3ec861f ... 26fa4cce66

4 changed files with 218 additions and 1197 deletions
Show stats Expand all files Collapse all files

4
.gitignore vendored
View file

@ -15,6 +15,4 @@ Testing/
*.pdb
.DS_Store
tmp/
.amduat-asl
.codex
.claude
.amduat-asl

21
include/amduat/asl/asl_store_index_fs.h
View file

@ -13,10 +13,6 @@ extern "C" {
#endif
enum { AMDUAT_ASL_STORE_INDEX_FS_ROOT_MAX = 1024 };
enum {
AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_CAP = 256,
AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_DIGEST_MAX = 64
};
typedef struct {
bool enabled;
@ -36,15 +32,6 @@ typedef struct {
uint8_t record_visibility;
} amduat_asl_store_index_fs_visibility_policy_t;
typedef struct {
bool valid;
amduat_asl_snapshot_id_t snapshot_id;
amduat_asl_log_position_t log_position;
amduat_hash_id_t hash_id;
uint16_t digest_len;
uint8_t digest[AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_DIGEST_MAX];
} amduat_asl_store_index_fs_neg_cache_entry_t;
typedef struct {
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_snapshot_policy_t snapshot_policy;
@ -56,15 +43,7 @@ typedef struct {
uint64_t last_ingest_time_ns;
amduat_asl_snapshot_id_t next_snapshot_id;
bool snapshot_state_initialized;
bool log_tail_initialized;
uint64_t log_tail_logseq;
uint8_t log_tail_hash[32];
bool skip_exists_precheck;
void *open_segments;
amduat_asl_store_index_fs_neg_cache_entry_t
neg_cache[AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_CAP];
uint16_t neg_cache_next;
pthread_mutex_t neg_cache_mutex;
pthread_mutex_t write_mutex;
uint32_t write_depth;
int write_lock_fd;

817
src/adapters/asl_store_index_fs/asl_store_index_fs.c
View file

@ -138,123 +138,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_impl(
amduat_reference_t *out_ref,
amduat_asl_index_state_t *out_state);
static bool amduat_asl_store_index_fs_neg_cache_contains(
amduat_asl_store_index_fs_t *fs,
amduat_asl_index_state_t state,
amduat_reference_t ref) {
size_t i;
if (fs == NULL) {
return false;
}
if (ref.digest.len > AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_DIGEST_MAX ||
(ref.digest.len != 0u && ref.digest.data == NULL)) {
return false;
}
if (pthread_mutex_lock(&fs->neg_cache_mutex) != 0) {
return false;
}
for (i = 0u; i < AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_CAP; ++i) {
const amduat_asl_store_index_fs_neg_cache_entry_t *entry =
&fs->neg_cache[i];
if (!entry->valid) {
continue;
}
if (entry->snapshot_id != state.snapshot_id ||
entry->log_position != state.log_position ||
entry->hash_id != ref.hash_id ||
entry->digest_len != ref.digest.len) {
continue;
}
if (entry->digest_len == 0u ||
memcmp(entry->digest, ref.digest.data, entry->digest_len) == 0) {
pthread_mutex_unlock(&fs->neg_cache_mutex);
return true;
}
}
pthread_mutex_unlock(&fs->neg_cache_mutex);
return false;
}
static void amduat_asl_store_index_fs_neg_cache_store_miss(
amduat_asl_store_index_fs_t *fs,
amduat_asl_index_state_t state,
amduat_reference_t ref) {
amduat_asl_store_index_fs_neg_cache_entry_t *entry;
if (fs == NULL) {
return;
}
if (ref.digest.len > AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_DIGEST_MAX ||
(ref.digest.len != 0u && ref.digest.data == NULL)) {
return;
}
if (pthread_mutex_lock(&fs->neg_cache_mutex) != 0) {
return;
}
entry = &fs->neg_cache[fs->neg_cache_next];
fs->neg_cache_next =
(uint16_t)((fs->neg_cache_next + 1u) %
AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_CAP);
entry->valid = true;
entry->snapshot_id = state.snapshot_id;
entry->log_position = state.log_position;
entry->hash_id = ref.hash_id;
entry->digest_len = (uint16_t)ref.digest.len;
if (entry->digest_len != 0u) {
memcpy(entry->digest, ref.digest.data, entry->digest_len);
}
pthread_mutex_unlock(&fs->neg_cache_mutex);
}
static void amduat_asl_store_index_fs_neg_cache_invalidate_ref(
amduat_asl_store_index_fs_t *fs,
amduat_reference_t ref) {
size_t i;
if (fs == NULL) {
return;
}
if (ref.digest.len > AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_DIGEST_MAX ||
(ref.digest.len != 0u && ref.digest.data == NULL)) {
return;
}
if (pthread_mutex_lock(&fs->neg_cache_mutex) != 0) {
return;
}
for (i = 0u; i < AMDUAT_ASL_STORE_INDEX_FS_NEG_CACHE_CAP; ++i) {
amduat_asl_store_index_fs_neg_cache_entry_t *entry = &fs->neg_cache[i];
if (!entry->valid) {
continue;
}
if (entry->hash_id != ref.hash_id || entry->digest_len != ref.digest.len) {
continue;
}
if (entry->digest_len == 0u ||
memcmp(entry->digest, ref.digest.data, entry->digest_len) == 0) {
entry->valid = false;
}
}
pthread_mutex_unlock(&fs->neg_cache_mutex);
}
static bool amduat_asl_store_index_fs_env_truthy(const char *value) {
if (value == NULL || value[0] == '\0') {
return false;
}
if (strcmp(value, "1") == 0 || strcmp(value, "true") == 0 ||
strcmp(value, "TRUE") == 0 || strcmp(value, "yes") == 0 ||
strcmp(value, "YES") == 0 || strcmp(value, "on") == 0 ||
strcmp(value, "ON") == 0) {
return true;
}
return false;
}
static int amduat_asl_store_index_fs_lockfile_acquire(
amduat_asl_store_index_fs_t *fs) {
char lock_path[AMDUAT_ASL_STORE_INDEX_FS_ROOT_MAX + 32];
@ -311,8 +194,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_begin_write(
return AMDUAT_ASL_STORE_ERR_IO;
}
fs->write_lock_fd = lock_fd;
/* Another store instance may have appended while we were unlocked. */
fs->log_tail_initialized = false;
}
fs->write_depth += 1u;
return AMDUAT_ASL_STORE_OK;
@ -1312,6 +1193,28 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_load_log(
return AMDUAT_ASL_STORE_OK;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_write_log(
const char *log_path,
const char *log_dir,
amduat_asl_log_record_t *records,
size_t record_count) {
amduat_octets_t log_bytes;
amduat_asl_store_index_fs_write_status_t status;
if (!amduat_enc_asl_log_encode_v1(records, record_count, &log_bytes)) {
return AMDUAT_ASL_STORE_ERR_IO;
}
status = amduat_asl_store_index_fs_write_replace(log_dir, log_path,
log_bytes.data,
log_bytes.len);
amduat_octets_free(&log_bytes);
if (status == AMDUAT_ASL_STORE_INDEX_FS_WRITE_OK) {
return AMDUAT_ASL_STORE_OK;
}
return AMDUAT_ASL_STORE_ERR_IO;
}
static bool amduat_asl_store_index_fs_encode_artifact_ref(
amduat_reference_t ref,
uint8_t *out,
@ -1347,240 +1250,6 @@ static bool amduat_asl_store_index_fs_encode_artifact_ref(
return true;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_load_log_tail(
const char *log_path,
uint64_t *out_last_logseq,
uint8_t out_last_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN]) {
int fd;
uint8_t *header_bytes = NULL;
size_t header_size = 0u;
uint64_t last_logseq = 0u;
uint8_t prev_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
bool seen_record = false;
if (log_path == NULL || out_last_logseq == NULL || out_last_hash == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
}
memset(prev_hash, 0, sizeof(prev_hash));
fd = open(log_path, O_RDONLY);
if (fd < 0) {
if (errno == ENOENT || errno == ENOTDIR) {
*out_last_logseq = 0u;
memcpy(out_last_hash, prev_hash, sizeof(prev_hash));
return AMDUAT_ASL_STORE_OK;
}
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_log_read_header_bytes(fd,
&header_bytes,
&header_size)) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
free(header_bytes);
(void)header_size;
while (true) {
uint8_t record_header[16];
uint64_t logseq;
uint32_t record_type;
uint32_t payload_len;
uint8_t *payload = NULL;
uint8_t record_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
uint8_t expected_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
int rc;
rc = amduat_asl_store_index_fs_log_read_exact_eof(fd,
record_header,
sizeof(record_header));
if (rc == 0) {
break;
}
if (rc < 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
logseq = amduat_asl_store_index_fs_load_u64_le(record_header);
if (seen_record && logseq <= last_logseq) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
record_type =
amduat_asl_store_index_fs_log_load_u32_le(record_header + 8u);
payload_len =
amduat_asl_store_index_fs_log_load_u32_le(record_header + 12u);
if (payload_len != 0u) {
payload = (uint8_t *)malloc(payload_len);
if (payload == NULL) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_log_read_exact(fd,
payload,
payload_len)) {
free(payload);
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
}
if (!amduat_asl_store_index_fs_log_read_exact(fd,
record_hash,
sizeof(record_hash))) {
free(payload);
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
if (!amduat_asl_store_index_fs_log_hash_record(prev_hash,
logseq,
record_type,
payload_len,
payload,
expected_hash) ||
memcmp(expected_hash, record_hash, sizeof(record_hash)) != 0) {
free(payload);
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
memcpy(prev_hash, record_hash, sizeof(prev_hash));
last_logseq = logseq;
seen_record = true;
free(payload);
}
close(fd);
*out_last_logseq = last_logseq;
memcpy(out_last_hash, prev_hash, sizeof(prev_hash));
return AMDUAT_ASL_STORE_OK;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_file(
const char *index_path,
const char *log_path,
uint64_t prev_logseq,
const uint8_t prev_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN],
uint32_t record_type,
const uint8_t *payload,
size_t payload_len,
uint64_t *out_new_logseq,
uint8_t out_new_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN]) {
int fd = -1;
struct stat st;
bool needs_header = false;
uint8_t *header_bytes = NULL;
size_t header_size = 0u;
uint8_t record_header[16];
uint8_t record_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
uint64_t new_logseq;
if (index_path == NULL || log_path == NULL || prev_hash == NULL ||
out_new_logseq == NULL || out_new_hash == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (payload_len != 0u && payload == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (payload_len > UINT32_MAX || prev_logseq == UINT64_MAX) {
return AMDUAT_ASL_STORE_ERR_UNSUPPORTED;
}
new_logseq = prev_logseq + 1u;
fd = open(log_path, O_RDWR | O_CREAT, 0644);
if (fd < 0) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (fstat(fd, &st) != 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
needs_header = st.st_size == 0;
if (!needs_header) {
if (lseek(fd, 0, SEEK_SET) < 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_log_read_header_bytes(fd,
&header_bytes,
&header_size)) {
close(fd);
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
free(header_bytes);
(void)header_size;
}
if (lseek(fd, 0, SEEK_END) < 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (needs_header) {
uint8_t header[AMDUAT_ASL_STORE_INDEX_FS_LOG_HEADER_LEN];
memcpy(header,
k_amduat_asl_store_index_fs_log_magic,
AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN);
amduat_asl_store_index_fs_log_store_u32_le(
header + AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN,
AMDUAT_ASL_STORE_INDEX_FS_LOG_VERSION);
amduat_asl_store_index_fs_log_store_u32_le(
header + AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN + 4u,
AMDUAT_ASL_STORE_INDEX_FS_LOG_HEADER_LEN);
amduat_asl_store_index_fs_log_store_u64_le(
header + AMDUAT_ASL_STORE_INDEX_FS_LOG_MAGIC_LEN + 8u,
0u);
if (!amduat_asl_store_index_fs_log_write_exact(fd, header, sizeof(header))) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
}
if (!amduat_asl_store_index_fs_log_hash_record(prev_hash,
new_logseq,
record_type,
(uint32_t)payload_len,
payload,
record_hash)) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
amduat_asl_store_index_fs_log_store_u64_le(record_header, new_logseq);
amduat_asl_store_index_fs_log_store_u32_le(record_header + 8u, record_type);
amduat_asl_store_index_fs_log_store_u32_le(record_header + 12u,
(uint32_t)payload_len);
if (!amduat_asl_store_index_fs_log_write_exact(fd,
record_header,
sizeof(record_header)) ||
(payload_len != 0u &&
!amduat_asl_store_index_fs_log_write_exact(fd, payload, payload_len)) ||
!amduat_asl_store_index_fs_log_write_exact(fd,
record_hash,
sizeof(record_hash))) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (fsync(fd) != 0) {
close(fd);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (close(fd) != 0) {
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_fsync_directory(index_path)) {
return AMDUAT_ASL_STORE_ERR_IO;
}
*out_new_logseq = new_logseq;
memcpy(out_new_hash, record_hash, sizeof(record_hash));
return AMDUAT_ASL_STORE_OK;
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_unlocked(
amduat_asl_store_index_fs_t *fs,
uint32_t record_type,
@ -1589,8 +1258,11 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_unlo
amduat_asl_index_state_t *out_state) {
char *index_path = NULL;
char *log_path = NULL;
uint64_t new_logseq = 0u;
uint8_t new_hash[AMDUAT_ASL_STORE_INDEX_FS_LOG_HASH_LEN];
amduat_asl_log_record_t *log_records = NULL;
size_t log_count = 0u;
amduat_asl_log_record_t *next = NULL;
uint64_t new_logseq;
uint8_t *payload_copy = NULL;
amduat_asl_store_error_t err;
if (fs == NULL) {
@ -1615,38 +1287,74 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_append_log_record_unlo
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!fs->log_tail_initialized) {
err = amduat_asl_store_index_fs_load_log_tail(log_path,
&fs->log_tail_logseq,
fs->log_tail_hash);
if (err != AMDUAT_ASL_STORE_OK) {
free(log_path);
free(index_path);
return err;
}
fs->log_tail_initialized = true;
err = amduat_asl_store_index_fs_load_log(log_path, &log_records, &log_count);
if (err != AMDUAT_ASL_STORE_OK) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return err;
}
if (log_count != 0u && log_records == NULL) {
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
err = amduat_asl_store_index_fs_append_log_record_file(
index_path,
log_path,
fs->log_tail_logseq,
fs->log_tail_hash,
record_type,
payload,
payload_len,
&new_logseq,
new_hash);
if (log_count == 0u) {
log_records = (amduat_asl_log_record_t *)calloc(1u,
sizeof(*log_records));
if (log_records == NULL) {
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
new_logseq = 1u;
} else {
if (log_records[log_count - 1u].logseq == UINT64_MAX) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
new_logseq = log_records[log_count - 1u].logseq + 1u;
next = (amduat_asl_log_record_t *)realloc(
log_records, (log_count + 1u) * sizeof(*log_records));
if (next == NULL) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
log_records = next;
}
if (payload_len != 0u) {
payload_copy = (uint8_t *)malloc(payload_len);
if (payload_copy == NULL) {
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
memcpy(payload_copy, payload, payload_len);
}
log_records[log_count].logseq = new_logseq;
log_records[log_count].record_type = record_type;
log_records[log_count].payload = amduat_octets(payload_copy, payload_len);
memset(log_records[log_count].record_hash, 0,
sizeof(log_records[log_count].record_hash));
log_count += 1u;
err = amduat_asl_store_index_fs_write_log(log_path, index_path,
log_records, log_count);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
if (err != AMDUAT_ASL_STORE_OK) {
return err;
}
fs->log_tail_logseq = new_logseq;
memcpy(fs->log_tail_hash, new_hash, sizeof(new_hash));
fs->log_tail_initialized = true;
if (out_state != NULL &&
!amduat_asl_store_index_fs_current_state_impl(fs, out_state)) {
return AMDUAT_ASL_STORE_ERR_IO;
@ -3698,6 +3406,7 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_load_segment(
uint8_t *segment_bytes;
size_t segment_len;
amduat_asl_store_index_fs_read_status_t status;
bool has_summary_hash;
if (out_segment == NULL || out_hash == NULL) {
return AMDUAT_ASL_STORE_ERR_IO;
@ -3722,17 +3431,16 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_load_segment(
return AMDUAT_ASL_STORE_ERR_IO;
}
/*
* Compute segment hash directly from bytes. Summary lookup performs a
* linear scan over 40-byte entries and is invoked per segment load, which
* can degenerate into extremely slow O(N^2) reads on large indexes.
*/
if (!amduat_hash_asl1_digest(AMDUAT_HASH_ASL1_ID_SHA256,
amduat_octets(segment_bytes, segment_len),
out_hash,
AMDUAT_ASL_STORE_INDEX_FS_SEGMENT_HASH_LEN)) {
free(segment_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
has_summary_hash = amduat_asl_store_index_fs_summary_lookup_hash(
fs, shard_id, segment_id, out_hash);
if (!has_summary_hash) {
if (!amduat_hash_asl1_digest(AMDUAT_HASH_ASL1_ID_SHA256,
amduat_octets(segment_bytes, segment_len),
out_hash,
AMDUAT_ASL_STORE_INDEX_FS_SEGMENT_HASH_LEN)) {
free(segment_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
}
if (!amduat_enc_asl_core_index_decode_v1(
@ -3955,104 +3663,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_materialize_artifact(
return AMDUAT_ASL_STORE_OK;
}
/*
* Bloom precheck: read only the segment header + bloom filter bytes (at most
* AMDUAT_ASL_CORE_INDEX_HEADER_SIZE + AMDUAT_ASL_INDEX_BLOOM_BYTES = 368
* bytes) and test the bloom filter before committing to a full segment load.
*
* Returns false only when the bloom filter definitively excludes ref, meaning
* the segment cannot contain it. Returns true in all other cases (bloom hit,
* no bloom stored, short read, bad magic) so the caller falls through to the
* full load path, which handles errors properly.
*
* Bloom filters have no false negatives, so returning false here is always
* safe: it can only cause a miss, never a spurious NOT_FOUND on a present ref.
*/
static bool amduat_asl_store_index_fs_bloom_precheck(const char *segment_path,
amduat_reference_t ref) {
enum {
k_precheck_buf_size =
AMDUAT_ASL_CORE_INDEX_HEADER_SIZE + AMDUAT_ASL_INDEX_BLOOM_BYTES
};
uint8_t buf[k_precheck_buf_size];
size_t total;
uint64_t magic;
uint64_t bloom_offset;
uint64_t bloom_size;
amduat_octets_t bloom;
int fd;
if (segment_path == NULL) {
return true;
}
fd = open(segment_path, O_RDONLY);
if (fd < 0) {
return true;
}
total = 0u;
while (total < sizeof(buf)) {
ssize_t n = read(fd, buf + total, sizeof(buf) - total);
if (n < 0) {
if (errno == EINTR) {
continue;
}
close(fd);
return true;
}
if (n == 0) {
break;
}
total += (size_t)n;
}
close(fd);
if (total < (size_t)AMDUAT_ASL_CORE_INDEX_HEADER_SIZE) {
return true;
}
/* Validate magic before trusting any other header field. */
magic = (uint64_t)buf[0] | ((uint64_t)buf[1] << 8) |
((uint64_t)buf[2] << 16) | ((uint64_t)buf[3] << 24) |
((uint64_t)buf[4] << 32) | ((uint64_t)buf[5] << 40) |
((uint64_t)buf[6] << 48) | ((uint64_t)buf[7] << 56);
if (magic != 0x33305844494c5341ull) {
return true;
}
/* bloom_offset is at header byte 48; bloom_size at header byte 56. */
bloom_offset = (uint64_t)buf[48] | ((uint64_t)buf[49] << 8) |
((uint64_t)buf[50] << 16) | ((uint64_t)buf[51] << 24) |
((uint64_t)buf[52] << 32) | ((uint64_t)buf[53] << 40) |
((uint64_t)buf[54] << 48) | ((uint64_t)buf[55] << 56);
bloom_size = (uint64_t)buf[56] | ((uint64_t)buf[57] << 8) |
((uint64_t)buf[58] << 16) | ((uint64_t)buf[59] << 24) |
((uint64_t)buf[60] << 32) | ((uint64_t)buf[61] << 40) |
((uint64_t)buf[62] << 48) | ((uint64_t)buf[63] << 56);
if (bloom_size == 0u) {
/* Segment has no bloom; cannot prefilter. */
return true;
}
if (bloom_offset != (uint64_t)AMDUAT_ASL_CORE_INDEX_HEADER_SIZE) {
/* Unexpected layout; fall through to full load. */
return true;
}
if (bloom_size > (uint64_t)AMDUAT_ASL_INDEX_BLOOM_BYTES) {
/* Bloom larger than precheck buffer; fall through. */
return true;
}
if (total < (size_t)AMDUAT_ASL_CORE_INDEX_HEADER_SIZE + (size_t)bloom_size) {
/* Short read; fall through. */
return true;
}
bloom = amduat_octets(buf + AMDUAT_ASL_CORE_INDEX_HEADER_SIZE,
(size_t)bloom_size);
return amduat_asl_index_bloom_maybe_contains(bloom, ref.hash_id, ref.digest);
}
static amduat_asl_store_error_t amduat_asl_store_index_fs_scan_segments(
amduat_asl_store_index_fs_t *fs,
const amduat_asl_replay_state_t *replay_state,
@ -4100,19 +3710,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_scan_segments(
return AMDUAT_ASL_STORE_ERR_IO;
}
/*
* Precheck the bloom filter by reading only the segment header and bloom
* bytes (≤368 bytes). Bloom filters have no false negatives, so if the
* precheck returns false the ref cannot be in this segment and we skip
* the full load (stat, malloc, read-all, SHA256, decode). On any I/O or
* parse error the precheck returns true and we fall through to the full
* load path, which surfaces the error properly.
*/
if (!amduat_asl_store_index_fs_bloom_precheck(segment_path, ref)) {
free(segment_path);
continue;
}
err = amduat_asl_store_index_fs_load_segment(fs,
shard_id,
seal->segment_id,
@ -4303,35 +3900,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_get_indexed_impl(
(ref.digest.len != 0u && ref.digest.data == NULL)) {
return AMDUAT_ASL_STORE_ERR_INTEGRITY;
}
if (amduat_asl_store_index_fs_neg_cache_contains(fs, state, ref)) {
return AMDUAT_ASL_STORE_ERR_NOT_FOUND;
}
if (state.snapshot_id != 0u) {
/*
* Fast path: when the requested state is exactly a snapshot anchor,
* snapshot metadata is already a complete replay state.
*/
if (amduat_asl_store_index_fs_load_snapshot_replay(fs->root_path,
state.snapshot_id,
NULL,
&replay_state,
&anchor_logseq)) {
if (replay_state.state.log_position == anchor_logseq &&
state.log_position == anchor_logseq) {
err = amduat_asl_store_index_fs_scan_segments(fs,
&replay_state,
ref,
out_artifact);
amduat_asl_replay_free(&replay_state);
if (err == AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
amduat_asl_store_index_fs_neg_cache_store_miss(fs, state, ref);
}
return err;
}
amduat_asl_replay_free(&replay_state);
}
}
if (!amduat_asl_store_index_fs_layout_build_log_path(fs->root_path,
&log_path)) {
@ -4388,9 +3956,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_get_indexed_impl(
out_artifact);
amduat_enc_asl_log_free(records, record_count);
amduat_asl_replay_free(&replay_state);
if (err == AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
amduat_asl_store_index_fs_neg_cache_store_miss(fs, state, ref);
}
return err;
}
@ -4514,9 +4079,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_tombstone_lift_impl(
payload_len,
out_state);
free(payload);
if (err == AMDUAT_ASL_STORE_OK) {
amduat_asl_store_index_fs_neg_cache_invalidate_ref(fs, ref);
}
return err;
}
@ -4552,7 +4114,6 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
amduat_reference_t derived_ref;
amduat_octets_t artifact_bytes;
amduat_asl_index_state_t current_state;
bool has_current_state = false;
amduat_artifact_t existing_artifact;
amduat_asl_store_error_t err;
char *index_path;
@ -4560,6 +4121,7 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
char *blocks_path;
char *segment_path;
char *block_path;
char *log_path;
uint16_t shard_id;
uint64_t segment_id;
uint64_t local_segment_id;
@ -4568,8 +4130,11 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
amduat_asl_extent_record_t extent;
amduat_octets_t segment_bytes;
uint8_t segment_hash[AMDUAT_ASL_STORE_INDEX_FS_SEGMENT_HASH_LEN];
amduat_asl_log_record_t *log_records;
size_t log_count;
uint8_t *seal_payload;
size_t seal_len;
uint64_t new_logseq;
amduat_asl_store_index_fs_write_status_t write_status;
size_t artifact_len;
@ -4598,33 +4163,29 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
fs->shard_count);
}
if (!fs->skip_exists_precheck) {
if (!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
has_current_state = true;
if (!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
existing_artifact = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_index_fs_get_indexed_impl(ctx, derived_ref,
current_state,
&existing_artifact);
if (err == AMDUAT_ASL_STORE_OK) {
amduat_artifact_free(&existing_artifact);
amduat_asl_store_index_fs_neg_cache_invalidate_ref(fs, derived_ref);
*out_ref = derived_ref;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
current_state.log_position);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_OK;
}
if (err != AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return err;
}
existing_artifact = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_index_fs_get_indexed_impl(ctx, derived_ref,
current_state,
&existing_artifact);
if (err == AMDUAT_ASL_STORE_OK) {
amduat_artifact_free(&existing_artifact);
*out_ref = derived_ref;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
current_state.log_position);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_OK;
}
if (err != AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return err;
}
if (fs->segment_policy.small_artifact_threshold != 0u &&
@ -4643,14 +4204,7 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
if (open_segment->active &&
amduat_asl_store_index_fs_open_segment_contains(open_segment,
derived_ref)) {
amduat_asl_store_index_fs_neg_cache_invalidate_ref(fs, derived_ref);
*out_ref = derived_ref;
if (!has_current_state &&
!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
has_current_state = true;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
current_state.log_position);
@ -4700,20 +4254,12 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
return err;
}
} else {
if (!has_current_state &&
!amduat_asl_store_index_fs_current_state_impl(ctx, &current_state)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
return AMDUAT_ASL_STORE_ERR_IO;
}
has_current_state = true;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
current_state.log_position);
}
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_neg_cache_invalidate_ref(fs, derived_ref);
*out_ref = derived_ref;
amduat_asl_store_index_fs_maybe_snapshot_size(fs);
return AMDUAT_ASL_STORE_OK;
@ -4924,11 +4470,75 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
return AMDUAT_ASL_STORE_ERR_IO;
}
if (!amduat_asl_store_index_fs_layout_build_log_path(fs->root_path,
&log_path)) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
log_records = NULL;
log_count = 0u;
err = amduat_asl_store_index_fs_load_log(log_path, &log_records, &log_count);
if (err != AMDUAT_ASL_STORE_OK) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return err;
}
if (log_count != 0u && log_records == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
if (log_count == 0u) {
log_records = (amduat_asl_log_record_t *)calloc(1u,
sizeof(*log_records));
if (log_records == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
} else {
amduat_asl_log_record_t *next = (amduat_asl_log_record_t *)realloc(
log_records, (log_count + 1u) * sizeof(*log_records));
if (next == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_enc_asl_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
return AMDUAT_ASL_STORE_ERR_IO;
}
log_records = next;
}
seal_len = 8u + AMDUAT_ASL_STORE_INDEX_FS_SEGMENT_HASH_LEN;
seal_payload = (uint8_t *)malloc(seal_len);
if (seal_payload == NULL) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
@ -4944,13 +4554,34 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
seal_payload[6] = (uint8_t)((segment_id >> 48) & 0xffu);
seal_payload[7] = (uint8_t)((segment_id >> 56) & 0xffu);
memcpy(seal_payload + 8, segment_hash, sizeof(segment_hash));
err = amduat_asl_store_index_fs_append_log_record_unlocked(
fs,
AMDUAT_ASL_LOG_RECORD_SEGMENT_SEAL,
seal_payload,
seal_len,
NULL);
free(seal_payload);
if (log_count == 0u) {
new_logseq = 1u;
} else {
if (log_records[log_count - 1u].logseq == UINT64_MAX) {
amduat_reference_free(&derived_ref);
amduat_octets_free(&artifact_bytes);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
free(seal_payload);
return AMDUAT_ASL_STORE_ERR_IO;
}
new_logseq = log_records[log_count - 1u].logseq + 1u;
}
log_records[log_count].logseq = new_logseq;
log_records[log_count].record_type = AMDUAT_ASL_LOG_RECORD_SEGMENT_SEAL;
log_records[log_count].payload = amduat_octets(seal_payload, seal_len);
memset(log_records[log_count].record_hash, 0,
sizeof(log_records[log_count].record_hash));
log_count += 1u;
err = amduat_asl_store_index_fs_write_log(log_path, index_path,
log_records, log_count);
amduat_asl_store_index_fs_log_free(log_records, log_count);
free(log_path);
free(index_path);
free(segments_path);
free(blocks_path);
@ -4962,11 +4593,10 @@ static amduat_asl_store_error_t amduat_asl_store_index_fs_put_indexed_unlocked_i
return err;
}
amduat_asl_store_index_fs_neg_cache_invalidate_ref(fs, derived_ref);
*out_ref = derived_ref;
amduat_asl_store_index_fs_fill_index_state(out_state,
current_state.snapshot_id,
fs->log_tail_logseq);
new_logseq);
amduat_asl_store_index_fs_update_ingest_state(fs, artifact_len);
amduat_asl_store_index_fs_maybe_snapshot_size(fs);
return AMDUAT_ASL_STORE_OK;
@ -5170,7 +4800,6 @@ amduat_asl_store_error_t amduat_asl_store_index_fs_snapshot_create(
free(manifest_path);
return err;
}
fs->log_tail_initialized = false;
{
amduat_asl_snapshot_manifest_t verify_manifest;
@ -5595,14 +5224,7 @@ bool amduat_asl_store_index_fs_init(amduat_asl_store_index_fs_t *fs,
fs->last_ingest_time_ns = 0u;
fs->next_snapshot_id = 0u;
fs->snapshot_state_initialized = false;
fs->log_tail_initialized = false;
fs->log_tail_logseq = 0u;
memset(fs->log_tail_hash, 0, sizeof(fs->log_tail_hash));
fs->skip_exists_precheck = amduat_asl_store_index_fs_env_truthy(
getenv("AMDUAT_ASL_INDEX_FS_SKIP_EXISTS_PRECHECK"));
fs->open_segments = NULL;
memset(fs->neg_cache, 0, sizeof(fs->neg_cache));
fs->neg_cache_next = 0u;
fs->write_depth = 0u;
fs->write_lock_fd = -1;
if (pthread_mutexattr_init(&mutex_attr) != 0) {
@ -5616,11 +5238,6 @@ bool amduat_asl_store_index_fs_init(amduat_asl_store_index_fs_t *fs,
pthread_mutexattr_destroy(&mutex_attr);
return false;
}
if (pthread_mutex_init(&fs->neg_cache_mutex, &mutex_attr) != 0) {
pthread_mutex_destroy(&fs->write_mutex);
pthread_mutexattr_destroy(&mutex_attr);
return false;
}
pthread_mutexattr_destroy(&mutex_attr);
return true;
}

573
tests/asl/test_asl_store_index_fs.c
View file

@ -156,24 +156,6 @@ static bool read_file(const char *path, uint8_t **out_bytes, size_t *out_len) {
return true;
}
static bool write_file(const char *path, const uint8_t *bytes, size_t len) {
FILE *fp;
size_t written;
if (path == NULL || (len != 0u && bytes == NULL)) {
return false;
}
fp = fopen(path, "wb");
if (fp == NULL) {
return false;
}
written = fwrite(bytes, 1u, len, fp);
if (fclose(fp) != 0) {
return false;
}
return written == len;
}
static bool build_log_path(const char *root, char **out_path) {
char *index_path = NULL;
bool ok;
@ -743,349 +725,6 @@ cleanup:
return exit_code;
}
static int test_snapshot_fast_path_with_corrupt_log(void) {
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_t fs;
amduat_asl_store_t store;
amduat_asl_store_error_t err;
amduat_asl_index_state_t state;
amduat_artifact_t artifact;
amduat_artifact_t loaded;
amduat_reference_t ref;
uint8_t payload[10];
char *root;
char *log_path = NULL;
static const uint8_t corrupt_log[] = {'B', 'A', 'D'};
int exit_code = 1;
root = make_temp_root();
if (root == NULL) {
fprintf(stderr, "temp root failed\n");
return 1;
}
memset(&config, 0, sizeof(config));
config.encoding_profile_id = AMDUAT_ENC_ASL1_CORE_V1;
config.hash_id = AMDUAT_HASH_ASL1_ID_SHA256;
if (!amduat_asl_store_index_fs_init(&fs, config, root)) {
fprintf(stderr, "index fs init failed\n");
goto cleanup;
}
amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);
memset(payload, 0x4c, sizeof(payload));
artifact = amduat_artifact(amduat_octets(payload, sizeof(payload)));
ref = amduat_reference(0u, amduat_octets(NULL, 0u));
err = amduat_asl_store_put_indexed(&store, artifact, &ref, &state);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "put_indexed failed: %d\n", err);
goto cleanup;
}
err = amduat_asl_store_index_fs_snapshot_create(&fs, 1u, NULL, NULL);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "snapshot create failed: %d\n", err);
goto cleanup;
}
if (!amduat_asl_index_current_state(&store, &state)) {
fprintf(stderr, "current_state failed\n");
goto cleanup;
}
if (state.snapshot_id != 1u) {
fprintf(stderr, "expected snapshot_id=1\n");
goto cleanup;
}
if (!build_log_path(root, &log_path) ||
!write_file(log_path, corrupt_log, sizeof(corrupt_log))) {
fprintf(stderr, "corrupt log write failed\n");
goto cleanup;
}
loaded = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_get_indexed(&store, ref, state, &loaded);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "get_indexed with snapshot fast path failed: %d\n", err);
goto cleanup;
}
if (!amduat_artifact_eq(artifact, loaded)) {
fprintf(stderr, "artifact mismatch from snapshot fast path\n");
amduat_artifact_free(&loaded);
goto cleanup;
}
amduat_artifact_free(&loaded);
exit_code = 0;
cleanup:
free(log_path);
amduat_reference_free(&ref);
if (!remove_tree(root)) {
fprintf(stderr, "cleanup failed\n");
exit_code = 1;
}
free(root);
return exit_code;
}
static int test_snapshot_fallback_replays_delta(void) {
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_t fs;
amduat_asl_store_t store;
amduat_asl_store_error_t err;
amduat_asl_index_state_t snapshot_state;
amduat_asl_index_state_t fallback_state;
amduat_artifact_t artifact_a;
amduat_artifact_t loaded;
amduat_reference_t ref_a;
uint8_t payload_a[7];
char *root;
int exit_code = 1;
root = make_temp_root();
if (root == NULL) {
fprintf(stderr, "temp root failed\n");
return 1;
}
memset(&config, 0, sizeof(config));
config.encoding_profile_id = AMDUAT_ENC_ASL1_CORE_V1;
config.hash_id = AMDUAT_HASH_ASL1_ID_SHA256;
if (!amduat_asl_store_index_fs_init(&fs, config, root)) {
fprintf(stderr, "index fs init failed\n");
goto cleanup;
}
amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);
memset(payload_a, 0x11, sizeof(payload_a));
artifact_a = amduat_artifact(amduat_octets(payload_a, sizeof(payload_a)));
ref_a = amduat_reference(0u, amduat_octets(NULL, 0u));
err = amduat_asl_store_put_indexed(&store, artifact_a, &ref_a, &snapshot_state);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "put_indexed a failed: %d\n", err);
goto cleanup;
}
err = amduat_asl_store_index_fs_snapshot_create(&fs, 1u, NULL, NULL);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "snapshot create failed: %d\n", err);
goto cleanup;
}
if (!amduat_asl_index_current_state(&store, &snapshot_state)) {
fprintf(stderr, "snapshot current_state failed\n");
goto cleanup;
}
if (snapshot_state.log_position == 0u) {
fprintf(stderr, "snapshot log position was zero\n");
goto cleanup;
}
fallback_state = snapshot_state;
fallback_state.log_position -= 1u;
loaded = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_get_indexed(&store, ref_a, fallback_state, &loaded);
if (err != AMDUAT_ASL_STORE_ERR_INTEGRITY) {
fprintf(stderr, "fallback expected integrity: %d\n", err);
amduat_artifact_free(&loaded);
goto cleanup;
}
amduat_artifact_free(&loaded);
exit_code = 0;
cleanup:
amduat_reference_free(&ref_a);
if (!remove_tree(root)) {
fprintf(stderr, "cleanup failed\n");
exit_code = 1;
}
free(root);
return exit_code;
}
static int test_snapshot_fast_path_not_found_and_tombstoned(void) {
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_t fs;
amduat_asl_store_t store;
amduat_asl_store_error_t err;
amduat_asl_index_state_t state;
amduat_asl_index_state_t tombstone_state;
amduat_reference_t ref;
amduat_reference_t missing_ref;
amduat_artifact_t artifact;
amduat_artifact_t loaded;
uint8_t payload[8];
uint8_t *missing_digest = NULL;
int exit_code = 1;
char *root;
root = make_temp_root();
if (root == NULL) {
fprintf(stderr, "temp root failed\n");
return 1;
}
memset(&config, 0, sizeof(config));
config.encoding_profile_id = AMDUAT_ENC_ASL1_CORE_V1;
config.hash_id = AMDUAT_HASH_ASL1_ID_SHA256;
if (!amduat_asl_store_index_fs_init(&fs, config, root)) {
fprintf(stderr, "index fs init failed\n");
goto cleanup;
}
amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);
memset(payload, 0x66, sizeof(payload));
artifact = amduat_artifact(amduat_octets(payload, sizeof(payload)));
ref = amduat_reference(0u, amduat_octets(NULL, 0u));
missing_ref = amduat_reference(0u, amduat_octets(NULL, 0u));
err = amduat_asl_store_put_indexed(&store, artifact, &ref, &state);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "put_indexed failed: %d\n", err);
goto cleanup;
}
err = amduat_asl_store_index_fs_snapshot_create(&fs, 1u, NULL, NULL);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "snapshot create failed: %d\n", err);
goto cleanup;
}
if (!amduat_asl_index_current_state(&store, &state)) {
fprintf(stderr, "current_state failed\n");
goto cleanup;
}
missing_digest = (uint8_t *)malloc(ref.digest.len);
if (missing_digest == NULL || ref.digest.len == 0u) {
fprintf(stderr, "missing ref build failed\n");
goto cleanup;
}
memcpy(missing_digest, ref.digest.data, ref.digest.len);
missing_digest[0] ^= 0xffu;
missing_ref.hash_id = ref.hash_id;
missing_ref.digest = amduat_octets(missing_digest, ref.digest.len);
loaded = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_get_indexed(&store, missing_ref, state, &loaded);
if (err != AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
fprintf(stderr, "snapshot not found expected: %d\n", err);
amduat_artifact_free(&loaded);
goto cleanup;
}
amduat_artifact_free(&loaded);
err = amduat_asl_store_tombstone(&store, ref, 1u, 2u, &tombstone_state);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "tombstone failed: %d\n", err);
goto cleanup;
}
err = amduat_asl_store_index_fs_snapshot_create(&fs, 2u, NULL, NULL);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "snapshot create after tombstone failed: %d\n", err);
goto cleanup;
}
if (!amduat_asl_index_current_state(&store, &state)) {
fprintf(stderr, "current_state after tombstone failed\n");
goto cleanup;
}
loaded = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_get_indexed(&store, ref, state, &loaded);
if (err != AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
fprintf(stderr, "snapshot tombstoned expected not found: %d\n", err);
amduat_artifact_free(&loaded);
goto cleanup;
}
amduat_artifact_free(&loaded);
exit_code = 0;
cleanup:
amduat_reference_free(&ref);
amduat_reference_free(&missing_ref);
if (!remove_tree(root)) {
fprintf(stderr, "cleanup failed\n");
exit_code = 1;
}
free(root);
return exit_code;
}
static int test_snapshot_fast_path_integrity_error_unchanged(void) {
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_t fs;
amduat_asl_store_t store;
amduat_asl_store_error_t err;
amduat_asl_index_state_t state;
amduat_artifact_t artifact;
amduat_artifact_t loaded;
amduat_reference_t ref;
uint8_t payload[6];
char *root;
char *segment_path = NULL;
static const uint8_t corrupt_segment[] = {'X', 'X', 'X'};
int exit_code = 1;
root = make_temp_root();
if (root == NULL) {
fprintf(stderr, "temp root failed\n");
return 1;
}
memset(&config, 0, sizeof(config));
config.encoding_profile_id = AMDUAT_ENC_ASL1_CORE_V1;
config.hash_id = AMDUAT_HASH_ASL1_ID_SHA256;
if (!amduat_asl_store_index_fs_init(&fs, config, root)) {
fprintf(stderr, "index fs init failed\n");
goto cleanup;
}
amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);
memset(payload, 0x31, sizeof(payload));
artifact = amduat_artifact(amduat_octets(payload, sizeof(payload)));
ref = amduat_reference(0u, amduat_octets(NULL, 0u));
err = amduat_asl_store_put_indexed(&store, artifact, &ref, &state);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "put_indexed failed: %d\n", err);
goto cleanup;
}
err = amduat_asl_store_index_fs_snapshot_create(&fs, 1u, NULL, NULL);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "snapshot create failed: %d\n", err);
goto cleanup;
}
if (!amduat_asl_index_current_state(&store, &state)) {
fprintf(stderr, "current_state failed\n");
goto cleanup;
}
if (!build_segment_path(root, 1u, &segment_path) ||
!write_file(segment_path, corrupt_segment, sizeof(corrupt_segment))) {
fprintf(stderr, "segment corruption failed\n");
goto cleanup;
}
loaded = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_get_indexed(&store, ref, state, &loaded);
if (err != AMDUAT_ASL_STORE_ERR_INTEGRITY) {
fprintf(stderr, "expected integrity error, got: %d\n", err);
amduat_artifact_free(&loaded);
goto cleanup;
}
amduat_artifact_free(&loaded);
exit_code = 0;
cleanup:
free(segment_path);
amduat_reference_free(&ref);
if (!remove_tree(root)) {
fprintf(stderr, "cleanup failed\n");
exit_code = 1;
}
free(root);
return exit_code;
}
static int test_segment_batch_packing(void) {
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_t fs;
@ -1955,219 +1594,10 @@ static int test_writer_reader_stress(void) {
return state.failed ? 1 : 0;
}
/*
* test_miss_latency_bounded
*
* Regression test for the miss-path bloom precheck optimisation.
*
* Builds a store with k_segment_count segments (one artifact each, sized at
* k_payload_bytes so each segment file is meaningfully large) then performs
* k_miss_trials GET lookups for a ref that is definitively absent. Each
* lookup must complete within k_miss_deadline_ns nanoseconds, and the total
* across all trials must be within k_total_deadline_ns.
*
* Without the bloom precheck every miss reads each candidate segment in full
* (stat + read + SHA256 + decode). With the precheck, only the 368-byte
* header+bloom prefix is read before the bloom filter short-circuits the
* lookup. On the test hardware this reduces per-miss latency by 10-20×;
* the bounds below are deliberately loose enough to pass on slow CI yet
* still catch a regression to the full-scan path.
*/
static int test_miss_latency_bounded(void) {
enum {
k_segment_count = 200,
k_miss_trials = 20,
k_payload_bytes = 16384
};
/* Each miss must finish within 2 s; total across all trials within 5 s.
* Pre-fix (full scan), 200 segments × 16 KB each ≈ 3.2 MB of I/O + SHA256
* per miss. Post-fix (bloom precheck) ≈ 73 KB per miss. Even on slow
* CI the 10× I/O reduction makes it straightforward to stay within 5 s. */
const uint64_t k_miss_deadline_ns = (uint64_t)2000 * 1000 * 1000;
const uint64_t k_total_deadline_ns = (uint64_t)5000 * 1000 * 1000;
amduat_asl_store_config_t config;
amduat_asl_store_index_fs_t fs;
amduat_asl_store_t store;
amduat_asl_store_error_t err;
amduat_asl_index_state_t state;
amduat_reference_t *refs = NULL;
uint8_t *payload = NULL;
uint8_t *missing_digest = NULL;
amduat_reference_t missing_ref;
char *root = NULL;
int exit_code = 1;
size_t i;
uint64_t total_miss_ns = 0u;
size_t digest_len = 0u;
missing_ref = amduat_reference(0u, amduat_octets(NULL, 0u));
root = make_temp_root();
if (root == NULL) {
fprintf(stderr, "test_miss_latency_bounded: temp root failed\n");
return 1;
}
memset(&config, 0, sizeof(config));
config.encoding_profile_id = AMDUAT_ENC_ASL1_CORE_V1;
config.hash_id = AMDUAT_HASH_ASL1_ID_SHA256;
if (!amduat_asl_store_index_fs_init(&fs, config, root)) {
fprintf(stderr, "test_miss_latency_bounded: index fs init failed\n");
goto cleanup;
}
amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);
refs = (amduat_reference_t *)calloc(k_segment_count, sizeof(*refs));
payload = (uint8_t *)malloc(k_payload_bytes);
if (refs == NULL || payload == NULL) {
fprintf(stderr, "test_miss_latency_bounded: alloc failed\n");
goto cleanup;
}
/* Fill payload with deterministic data. */
for (i = 0; i < (size_t)k_payload_bytes; ++i) {
payload[i] = (uint8_t)(i & 0xffu);
}
/* Insert k_segment_count artifacts, each in its own segment
* (default pending_bytes limit forces a flush every record). */
for (i = 0; i < (size_t)k_segment_count; ++i) {
/* Perturb payload per artifact so each has a distinct digest. */
payload[0] = (uint8_t)(i & 0xffu);
payload[1] = (uint8_t)((i >> 8) & 0xffu);
refs[i] = amduat_reference(0u, amduat_octets(NULL, 0u));
err = amduat_asl_store_put(&store,
amduat_artifact(amduat_octets(payload,
k_payload_bytes)),
&refs[i]);
if (err != AMDUAT_ASL_STORE_OK) {
fprintf(stderr, "test_miss_latency_bounded: put %zu failed: %d\n",
i, err);
goto cleanup;
}
}
if (!amduat_asl_index_current_state(&store, &state)) {
fprintf(stderr, "test_miss_latency_bounded: current_state failed\n");
goto cleanup;
}
/* Build a ref with a valid shape but a digest not present in the store.
* Flip the first byte of a known ref's digest. */
if (refs[0].digest.len == 0u || refs[0].digest.data == NULL) {
fprintf(stderr, "test_miss_latency_bounded: ref[0] has no digest\n");
goto cleanup;
}
digest_len = refs[0].digest.len;
missing_digest = (uint8_t *)malloc(digest_len);
if (missing_digest == NULL) {
fprintf(stderr, "test_miss_latency_bounded: missing_digest alloc failed\n");
goto cleanup;
}
memcpy(missing_digest, refs[0].digest.data, digest_len);
missing_digest[0] ^= 0xffu;
missing_ref = amduat_reference(refs[0].hash_id,
amduat_octets(missing_digest, digest_len));
/* Correctness check: the missing ref must return NOT_FOUND. */
{
amduat_artifact_t loaded = amduat_artifact(amduat_octets(NULL, 0u));
err = amduat_asl_store_get_indexed(&store, missing_ref, state, &loaded);
if (err != AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
fprintf(stderr,
"test_miss_latency_bounded: expected NOT_FOUND, got %d\n", err);
amduat_artifact_free(&loaded);
goto cleanup;
}
amduat_artifact_free(&loaded);
}
/* Performance: k_miss_trials miss GETs must each finish within deadline. */
for (i = 0; i < (size_t)k_miss_trials; ++i) {
amduat_artifact_t loaded = amduat_artifact(amduat_octets(NULL, 0u));
uint64_t t0 = now_ns();
uint64_t elapsed;
err = amduat_asl_store_get_indexed(&store, missing_ref, state, &loaded);
elapsed = now_ns() - t0;
amduat_artifact_free(&loaded);
if (err != AMDUAT_ASL_STORE_ERR_NOT_FOUND) {
fprintf(stderr,
"test_miss_latency_bounded: trial %zu: expected NOT_FOUND, "
"got %d\n",
i, err);
goto cleanup;
}
total_miss_ns += elapsed;
if (elapsed > k_miss_deadline_ns) {
fprintf(stderr,
"test_miss_latency_bounded: trial %zu took %.1f ms "
"(limit %.1f ms)\n",
i,
(double)elapsed / 1e6,
(double)k_miss_deadline_ns / 1e6);
goto cleanup;
}
}
fprintf(stderr,
"test_miss_latency_bounded: %d segments, %d miss trials, "
"total=%.1f ms, avg=%.2f ms/miss\n",
k_segment_count,
k_miss_trials,
(double)total_miss_ns / 1e6,
(double)total_miss_ns / 1e6 / (double)k_miss_trials);
if (total_miss_ns > k_total_deadline_ns) {
fprintf(stderr,
"test_miss_latency_bounded: total %.1f ms exceeds limit %.1f ms\n",
(double)total_miss_ns / 1e6,
(double)k_total_deadline_ns / 1e6);
goto cleanup;
}
exit_code = 0;
cleanup:
if (refs != NULL) {
for (i = 0; i < (size_t)k_segment_count; ++i) {
amduat_reference_free(&refs[i]);
}
}
free(refs);
free(payload);
/* missing_digest is owned by missing_ref; amduat_reference_free frees it. */
amduat_reference_free(&missing_ref);
if (root != NULL) {
if (!remove_tree(root)) {
fprintf(stderr, "test_miss_latency_bounded: cleanup failed\n");
exit_code = 1;
}
free(root);
}
return exit_code;
}
int main(void) {
if (test_round_trip() != 0) {
return 1;
}
if (test_snapshot_fast_path_with_corrupt_log() != 0) {
return 1;
}
if (test_snapshot_fallback_replays_delta() != 0) {
return 1;
}
if (test_snapshot_fast_path_not_found_and_tombstoned() != 0) {
return 1;
}
if (test_snapshot_fast_path_integrity_error_unchanged() != 0) {
return 1;
}
if (test_snapshot_truncation() != 0) {
return 1;
}
@ -2180,9 +1610,6 @@ int main(void) {
if (test_gc_keeps_visible() != 0) {
return 1;
}
if (test_miss_latency_bounded() != 0) {
return 1;
}
if (test_large_round_trip_perf() != 0) {
return 1;
}