amduat/tests/asl/test_asl_projection_replay.c

#include "amduat/asl/asl_store_index_fs.h"
#include "amduat/asl/log_cursor.h"
#include "amduat/asl/projection_checkpoint.h"
#include "amduat/asl/record.h"
#include "amduat/asl/store.h"
#include "amduat/enc/asl1_core.h"
#include "amduat/hash/asl1.h"

#include <dirent.h>
#include <errno.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

static char *make_temp_root(void) {
  char *templ;
  const char template_prefix[] = "/tmp/amduat_test_asl_projection_replay_XXXXXX";

  templ = (char *)malloc(sizeof(template_prefix));
  if (templ == NULL) {
    return NULL;
  }
  memcpy(templ, template_prefix, sizeof(template_prefix));
  if (mkdtemp(templ) == NULL) {
    free(templ);
    return NULL;
  }
  return templ;
}

static bool remove_tree(const char *path) {
  struct stat st;
  DIR *dir;
  struct dirent *entry;

  if (path == NULL) {
    return false;
  }
  if (stat(path, &st) != 0) {
    return errno == ENOENT;
  }
  if (!S_ISDIR(st.st_mode)) {
    return unlink(path) == 0;
  }

  dir = opendir(path);
  if (dir == NULL) {
    return false;
  }
  while ((entry = readdir(dir)) != NULL) {
    char child[2048];
    if (strcmp(entry->d_name, ".") == 0 ||
        strcmp(entry->d_name, "..") == 0) {
      continue;
    }
    snprintf(child, sizeof(child), "%s/%s", path, entry->d_name);
    if (!remove_tree(child)) {
      closedir(dir);
      return false;
    }
  }
  closedir(dir);
  return rmdir(path) == 0;
}

static bool records_equal(const amduat_asl_log_record_t *a,
                          const amduat_asl_log_record_t *b) {
  if (a->logseq != b->logseq || a->record_type != b->record_type) {
    return false;
  }
  if (a->payload.len != b->payload.len) {
    return false;
  }
  if (a->payload.len == 0u) {
    return true;
  }
  if (a->payload.data == NULL || b->payload.data == NULL) {
    return false;
  }
  return memcmp(a->payload.data, b->payload.data, a->payload.len) == 0;
}

static int test_cursor_deterministic_replay(void) {
  amduat_asl_store_config_t config;
  amduat_asl_store_index_fs_t fs;
  amduat_asl_store_t store;
  amduat_artifact_t artifact;
  amduat_reference_t ref;
  amduat_asl_index_state_t state;
  uint8_t payload_a[4] = {0x11, 0x22, 0x33, 0x44};
  uint8_t payload_b[4] = {0xaa, 0xbb, 0xcc, 0xdd};
  amduat_asl_log_record_t *all = NULL;
  size_t all_len = 0u;
  amduat_asl_log_position_t cursor = 0u;
  bool end = false;
  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_root;
  }
  amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);

  artifact = amduat_artifact(amduat_octets(payload_a, sizeof(payload_a)));
  ref = amduat_reference(0u, amduat_octets(NULL, 0u));
  if (amduat_asl_store_put_indexed(&store, artifact, &ref, &state) !=
      AMDUAT_ASL_STORE_OK) {
    fprintf(stderr, "put_indexed A failed\n");
    goto cleanup_root;
  }
  amduat_reference_free(&ref);

  artifact = amduat_artifact(amduat_octets(payload_b, sizeof(payload_b)));
  ref = amduat_reference(0u, amduat_octets(NULL, 0u));
  if (amduat_asl_store_put_indexed(&store, artifact, &ref, &state) !=
      AMDUAT_ASL_STORE_OK) {
    fprintf(stderr, "put_indexed B failed\n");
    goto cleanup_root;
  }
  amduat_reference_free(&ref);

  if (!amduat_asl_log_read_since(&store,
                                 0u,
                                 0u,
                                 &all,
                                 &all_len,
                                 &cursor,
                                 &end)) {
    fprintf(stderr, "log read all failed\n");
    goto cleanup_root;
  }

  cursor = 0u;
  for (size_t i = 0u; i < all_len; ++i) {
    amduat_asl_log_record_t *page = NULL;
    size_t page_len = 0u;
    amduat_asl_log_position_t next = 0u;
    bool page_end = false;

    if (!amduat_asl_log_read_since(&store,
                                   cursor,
                                   1u,
                                   &page,
                                   &page_len,
                                   &next,
                                   &page_end)) {
      fprintf(stderr, "log page read failed\n");
      goto cleanup_all;
    }
    if (page_len != 1u || !records_equal(&all[i], &page[0])) {
      amduat_enc_asl_log_free(page, page_len);
      fprintf(stderr, "log page mismatch\n");
      goto cleanup_all;
    }
    cursor = next;
    amduat_enc_asl_log_free(page, page_len);
  }

  {
    amduat_asl_log_record_t *page = NULL;
    size_t page_len = 0u;
    amduat_asl_log_position_t next = 0u;
    bool page_end = false;
    if (!amduat_asl_log_read_since(&store,
                                   cursor,
                                   1u,
                                   &page,
                                   &page_len,
                                   &next,
                                   &page_end)) {
      fprintf(stderr, "log tail read failed\n");
      goto cleanup_all;
    }
    if (page_len != 0u || !page_end) {
      amduat_enc_asl_log_free(page, page_len);
      fprintf(stderr, "log tail mismatch\n");
      goto cleanup_all;
    }
    amduat_enc_asl_log_free(page, page_len);
  }

  exit_code = 0;

cleanup_all:
  amduat_enc_asl_log_free(all, all_len);
cleanup_root:
  if (!remove_tree(root)) {
    fprintf(stderr, "cleanup failed\n");
    exit_code = 1;
  }
  free(root);
  return exit_code;
}

static int test_checkpoint_resume(void) {
  amduat_asl_store_config_t config;
  amduat_asl_store_index_fs_t fs;
  amduat_asl_store_t store;
  amduat_asl_index_state_t state;
  amduat_asl_projection_checkpoint_v1_t checkpoint;
  amduat_asl_projection_checkpoint_v1_t decoded;
  amduat_octets_t checkpoint_bytes = amduat_octets(NULL, 0u);
  amduat_reference_t checkpoint_ref;
  amduat_asl_record_t checkpoint_record;
  amduat_asl_log_record_t *page = NULL;
  size_t page_len = 0u;
  amduat_asl_log_position_t next = 0u;
  bool end = false;
  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_root;
  }
  amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);

  {
    uint8_t payload[3] = {0x01, 0x02, 0x03};
    amduat_artifact_t artifact = amduat_artifact(amduat_octets(payload, sizeof(payload)));
    if (amduat_asl_store_put_indexed(&store, artifact, &checkpoint_ref, &state) !=
        AMDUAT_ASL_STORE_OK) {
      fprintf(stderr, "put_indexed failed\n");
      goto cleanup_root;
    }
    amduat_reference_free(&checkpoint_ref);
  }

  if (!amduat_asl_index_current_state(&store, &state)) {
    fprintf(stderr, "current_state failed\n");
    goto cleanup_root;
  }

  memset(&checkpoint, 0, sizeof(checkpoint));
  checkpoint.checkpoint_version = AMDUAT_ASL_PROJECTION_CHECKPOINT_VERSION_V1;
  checkpoint.projection_name = amduat_octets("projection", strlen("projection"));
  checkpoint.projection_schema_version = 1u;
  checkpoint.has_snapshot_id = true;
  checkpoint.snapshot_id = state.snapshot_id;
  checkpoint.log_position = state.log_position;
  checkpoint.created_at = 0u;
  checkpoint.builder_id = amduat_octets("test", strlen("test"));

  if (!amduat_asl_projection_checkpoint_encode_v1(&checkpoint,
                                                  &checkpoint_bytes)) {
    fprintf(stderr, "checkpoint encode failed\n");
    goto cleanup_root;
  }

  if (amduat_asl_record_store_put(
          &store,
          amduat_octets("projection/checkpoint",
                        strlen("projection/checkpoint")),
          checkpoint_bytes,
          &checkpoint_ref) != AMDUAT_ASL_STORE_OK) {
    fprintf(stderr, "checkpoint record put failed\n");
    goto cleanup_bytes;
  }

  if (amduat_asl_record_store_get(&store, checkpoint_ref, &checkpoint_record) !=
      AMDUAT_ASL_STORE_OK) {
    fprintf(stderr, "checkpoint record get failed\n");
    goto cleanup_ref;
  }

  if (!amduat_asl_projection_checkpoint_decode_v1(
          checkpoint_record.payload, &decoded)) {
    fprintf(stderr, "checkpoint decode failed\n");
    amduat_asl_record_free(&checkpoint_record);
    goto cleanup_ref;
  }
  if (decoded.log_position != state.log_position) {
    fprintf(stderr, "checkpoint log position mismatch\n");
    amduat_asl_projection_checkpoint_free(&decoded);
    amduat_asl_record_free(&checkpoint_record);
    goto cleanup_ref;
  }
  amduat_asl_projection_checkpoint_free(&decoded);
  amduat_asl_record_free(&checkpoint_record);

  if (!amduat_asl_log_read_since(&store,
                                 state.log_position,
                                 0u,
                                 &page,
                                 &page_len,
                                 &next,
                                 &end)) {
    fprintf(stderr, "resume log read failed\n");
    goto cleanup_ref;
  }
  if (page_len == 0u) {
    fprintf(stderr, "resume expected non-empty log\n");
    amduat_enc_asl_log_free(page, page_len);
    goto cleanup_ref;
  }
  amduat_enc_asl_log_free(page, page_len);

  if (!amduat_asl_log_read_since(&store,
                                 next,
                                 0u,
                                 &page,
                                 &page_len,
                                 &next,
                                 &end)) {
    fprintf(stderr, "resume tail read failed\n");
    goto cleanup_ref;
  }
  if (page_len != 0u || !end) {
    fprintf(stderr, "resume expected empty tail\n");
    amduat_enc_asl_log_free(page, page_len);
    goto cleanup_ref;
  }
  amduat_enc_asl_log_free(page, page_len);

  exit_code = 0;

cleanup_ref:
  amduat_reference_free(&checkpoint_ref);
cleanup_bytes:
  amduat_octets_free(&checkpoint_bytes);
cleanup_root:
  if (!remove_tree(root)) {
    fprintf(stderr, "cleanup failed\n");
    exit_code = 1;
  }
  free(root);
  return exit_code;
}

static int test_projector_loop_stub(void) {
  amduat_asl_store_config_t config;
  amduat_asl_store_index_fs_t fs;
  amduat_asl_store_t store;
  amduat_asl_index_state_t state;
  amduat_asl_log_record_t *page = NULL;
  size_t page_len = 0u;
  amduat_asl_log_position_t after = 0u;
  amduat_asl_log_position_t next = 0u;
  bool end = false;
  amduat_reference_t checkpoint_ref;
  amduat_asl_record_t checkpoint_record;
  amduat_asl_projection_checkpoint_v1_t checkpoint;
  amduat_asl_projection_checkpoint_v1_t decoded;
  amduat_octets_t checkpoint_bytes = amduat_octets(NULL, 0u);
  size_t projected_count = 0u;
  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_root;
  }
  amduat_asl_store_init(&store, config, amduat_asl_store_index_fs_ops(), &fs);

  {
    uint8_t payload_a[2] = {0x10, 0x20};
    uint8_t payload_b[2] = {0x30, 0x40};
    amduat_reference_t ref = amduat_reference(0u, amduat_octets(NULL, 0u));
    amduat_artifact_t artifact = amduat_artifact(amduat_octets(payload_a, sizeof(payload_a)));
    if (amduat_asl_store_put_indexed(&store, artifact, &ref, &state) !=
        AMDUAT_ASL_STORE_OK) {
      fprintf(stderr, "put_indexed A failed\n");
      amduat_reference_free(&ref);
      goto cleanup_root;
    }
    amduat_reference_free(&ref);

    artifact = amduat_artifact(amduat_octets(payload_b, sizeof(payload_b)));
    ref = amduat_reference(0u, amduat_octets(NULL, 0u));
    if (amduat_asl_store_put_indexed(&store, artifact, &ref, &state) !=
        AMDUAT_ASL_STORE_OK) {
      fprintf(stderr, "put_indexed B failed\n");
      amduat_reference_free(&ref);
      goto cleanup_root;
    }
    amduat_reference_free(&ref);
  }

  if (!amduat_asl_index_current_state(&store, &state)) {
    fprintf(stderr, "current_state failed\n");
    goto cleanup_root;
  }

  checkpoint_ref = amduat_reference(0u, amduat_octets(NULL, 0u));
  while (!end) {
    if (!amduat_asl_log_read_since(&store,
                                   after,
                                   0u,
                                   &page,
                                   &page_len,
                                   &next,
                                   &end)) {
      fprintf(stderr, "projector log read failed\n");
      goto cleanup_checkpoint_ref;
    }

    projected_count += page_len;

    memset(&checkpoint, 0, sizeof(checkpoint));
    checkpoint.checkpoint_version = AMDUAT_ASL_PROJECTION_CHECKPOINT_VERSION_V1;
    checkpoint.projection_name =
        amduat_octets("projector-stub", strlen("projector-stub"));
    checkpoint.projection_schema_version = 1u;
    checkpoint.has_snapshot_id = true;
    checkpoint.snapshot_id = state.snapshot_id;
    checkpoint.log_position = next;
    checkpoint.created_at = 0u;
    checkpoint.builder_id =
        amduat_octets("test-projector", strlen("test-projector"));
    checkpoint.has_artifact_count = true;
    checkpoint.artifact_count = projected_count;

    if (!amduat_asl_projection_checkpoint_encode_v1(&checkpoint,
                                                    &checkpoint_bytes)) {
      fprintf(stderr, "checkpoint encode failed\n");
      amduat_enc_asl_log_free(page, page_len);
      goto cleanup_checkpoint_ref;
    }

    amduat_reference_free(&checkpoint_ref);
    if (amduat_asl_record_store_put(
            &store,
            amduat_octets("projection/checkpoint",
                          strlen("projection/checkpoint")),
            checkpoint_bytes,
            &checkpoint_ref) != AMDUAT_ASL_STORE_OK) {
      fprintf(stderr, "checkpoint store failed\n");
      amduat_octets_free(&checkpoint_bytes);
      amduat_enc_asl_log_free(page, page_len);
      goto cleanup_checkpoint_ref;
    }

    amduat_octets_free(&checkpoint_bytes);
    amduat_enc_asl_log_free(page, page_len);
    after = next;
  }

  if (amduat_asl_record_store_get(&store, checkpoint_ref, &checkpoint_record) !=
      AMDUAT_ASL_STORE_OK) {
    fprintf(stderr, "checkpoint record get failed\n");
    goto cleanup_checkpoint_ref;
  }

  if (!amduat_octets_eq(
          checkpoint_record.schema,
          amduat_octets("projection/checkpoint",
                        strlen("projection/checkpoint")))) {
    fprintf(stderr, "checkpoint schema mismatch\n");
    amduat_asl_record_free(&checkpoint_record);
    goto cleanup_checkpoint_ref;
  }

  if (!amduat_asl_projection_checkpoint_decode_v1(
          checkpoint_record.payload, &decoded)) {
    fprintf(stderr, "checkpoint decode failed\n");
    amduat_asl_record_free(&checkpoint_record);
    goto cleanup_checkpoint_ref;
  }

  if (decoded.log_position != state.log_position) {
    fprintf(stderr, "checkpoint log position mismatch\n");
    amduat_asl_projection_checkpoint_free(&decoded);
    amduat_asl_record_free(&checkpoint_record);
    goto cleanup_checkpoint_ref;
  }

  amduat_asl_projection_checkpoint_free(&decoded);
  amduat_asl_record_free(&checkpoint_record);
  exit_code = 0;

cleanup_checkpoint_ref:
  amduat_octets_free(&checkpoint_bytes);
  amduat_reference_free(&checkpoint_ref);
cleanup_root:
  if (!remove_tree(root)) {
    fprintf(stderr, "cleanup failed\n");
    exit_code = 1;
  }
  free(root);
  return exit_code;
}

int main(void) {
  if (test_cursor_deterministic_replay() != 0) {
    return 1;
  }
  if (test_checkpoint_resume() != 0) {
    return 1;
  }
  if (test_projector_loop_stub() != 0) {
    return 1;
  }
  return 0;
}