Add federation locations and conflict checks

include/amduat/fed/replay.h

@@ -31,9 +31,15 @@ typedef struct {
   amduat_reference_t ref;
 } amduat_fed_record_id_t;
 
+typedef struct {
+  uint32_t domain_id;
+  amduat_reference_t ref;
+} amduat_fed_record_location_t;
+
 typedef struct {
   amduat_fed_record_meta_t meta;
   amduat_fed_record_id_t id;
+  amduat_fed_record_location_t location;
   uint64_t logseq;
   uint64_t snapshot_id;
   uint64_t log_prefix;

include/amduat/fed/view.h

@@ -54,7 +54,15 @@ amduat_fed_resolve_error_t amduat_fed_resolve(
     const amduat_fed_view_t *view,
     amduat_asl_store_t *local_store,
     amduat_reference_t ref,
-    amduat_artifact_t *out_artifact);
+    amduat_artifact_t *out_artifact,
+    amduat_fed_record_location_t *out_location);
+
+amduat_fed_resolve_error_t amduat_fed_resolve_with_loc(
+    const amduat_fed_view_t *view,
+    amduat_asl_store_t *local_store,
+    amduat_reference_t ref,
+    amduat_artifact_t *out_artifact,
+    amduat_fed_record_location_t *out_location);
 
 #ifdef __cplusplus
 } /* extern "C" */

src/near_core/fed/ingest.c

@@ -16,6 +16,8 @@ static bool amduat_fed_record_equivalent(const amduat_fed_record_t *a,
          a->meta.visibility == b->meta.visibility &&
          a->meta.has_source == b->meta.has_source &&
          a->meta.source_domain == b->meta.source_domain &&
+         a->location.domain_id == b->location.domain_id &&
+         amduat_reference_eq(a->location.ref, b->location.ref) &&
          a->logseq == b->logseq &&
          a->snapshot_id == b->snapshot_id &&
          a->log_prefix == b->log_prefix;

src/near_core/fed/replay.c

@@ -53,6 +53,10 @@ static bool amduat_fed_record_clone(const amduat_fed_record_t *src,
   if (!amduat_reference_clone(src->id.ref, &out->id.ref)) {
     return false;
   }
+  if (!amduat_reference_clone(src->location.ref, &out->location.ref)) {
+    amduat_reference_free(&out->id.ref);
+    return false;
+  }
   return true;
 }
 
@@ -61,6 +65,7 @@ static void amduat_fed_record_free(amduat_fed_record_t *record) {
     return;
   }
   amduat_reference_free(&record->id.ref);
+  amduat_reference_free(&record->location.ref);
   memset(record, 0, sizeof(*record));
 }
 
@@ -150,6 +155,10 @@ bool amduat_fed_record_validate(const amduat_fed_record_t *record) {
       record->id.ref.digest.data == NULL) {
     return false;
   }
+  if (record->location.ref.digest.len != 0u &&
+      record->location.ref.digest.data == NULL) {
+    return false;
+  }
   return true;
 }
 

src/near_core/fed/view.c

@@ -12,6 +12,10 @@ static bool amduat_fed_record_clone(const amduat_fed_record_t *src,
   if (!amduat_reference_clone(src->id.ref, &out->id.ref)) {
     return false;
   }
+  if (!amduat_reference_clone(src->location.ref, &out->location.ref)) {
+    amduat_reference_free(&out->id.ref);
+    return false;
+  }
   return true;
 }
 
@@ -20,6 +24,7 @@ static void amduat_fed_record_free(amduat_fed_record_t *record) {
     return;
   }
   amduat_reference_free(&record->id.ref);
+  amduat_reference_free(&record->location.ref);
   memset(record, 0, sizeof(*record));
 }
 
@@ -31,6 +36,8 @@ static bool amduat_fed_record_equivalent(const amduat_fed_record_t *a,
          a->meta.source_domain == b->meta.source_domain &&
          a->id.type == b->id.type &&
          amduat_reference_eq(a->id.ref, b->id.ref) &&
+         a->location.domain_id == b->location.domain_id &&
+         amduat_reference_eq(a->location.ref, b->location.ref) &&
          a->logseq == b->logseq &&
          a->snapshot_id == b->snapshot_id &&
          a->log_prefix == b->log_prefix;
@@ -327,16 +334,30 @@ amduat_fed_resolve_error_t amduat_fed_resolve(
     const amduat_fed_view_t *view,
     amduat_asl_store_t *local_store,
     amduat_reference_t ref,
-    amduat_artifact_t *out_artifact) {
+    amduat_artifact_t *out_artifact,
+    amduat_fed_record_location_t *out_location) {
   amduat_asl_store_error_t store_err;
   const amduat_fed_record_t *record;
+  amduat_fed_record_location_t empty_location;
 
   if (local_store == NULL || out_artifact == NULL) {
     return AMDUAT_FED_RESOLVE_STORE_ERROR;
   }
 
+  if (out_location != NULL) {
+    empty_location.domain_id = 0u;
+    empty_location.ref = amduat_reference(0, amduat_octets(NULL, 0u));
+    *out_location = empty_location;
+  }
+
   store_err = amduat_asl_store_get(local_store, ref, out_artifact);
   if (store_err == AMDUAT_ASL_STORE_OK) {
+    if (out_location != NULL && view != NULL) {
+      out_location->domain_id = view->local_domain_id;
+      if (!amduat_reference_clone(ref, &out_location->ref)) {
+        out_location->ref = amduat_reference(0, amduat_octets(NULL, 0u));
+      }
+    }
     return AMDUAT_FED_RESOLVE_OK;
   }
   if (store_err == AMDUAT_ASL_STORE_ERR_INTEGRITY) {
@@ -355,7 +376,27 @@ amduat_fed_resolve_error_t amduat_fed_resolve(
     return AMDUAT_FED_RESOLVE_NOT_FOUND;
   }
   if (record->meta.domain_id != view->local_domain_id) {
+    if (out_location != NULL) {
+      out_location->domain_id = record->location.domain_id;
+      if (!amduat_reference_clone(record->location.ref,
+                                  &out_location->ref)) {
+        out_location->ref = amduat_reference(0, amduat_octets(NULL, 0u));
+      }
+    }
     return AMDUAT_FED_RESOLVE_FOUND_REMOTE_NO_BYTES;
   }
   return AMDUAT_FED_RESOLVE_NOT_FOUND;
 }
+
+amduat_fed_resolve_error_t amduat_fed_resolve_with_loc(
+    const amduat_fed_view_t *view,
+    amduat_asl_store_t *local_store,
+    amduat_reference_t ref,
+    amduat_artifact_t *out_artifact,
+    amduat_fed_record_location_t *out_location) {
+  return amduat_fed_resolve(view,
+                            local_store,
+                            ref,
+                            out_artifact,
+                            out_location);
+}

tests/fed/test_fed_ingest.c

@@ -21,6 +21,8 @@ static amduat_fed_record_t make_record(uint32_t domain_id,
   record.meta.has_source = 0;
   record.id.type = type;
   record.id.ref = ref;
+  record.location.domain_id = domain_id;
+  record.location.ref = ref;
   record.logseq = logseq;
   record.snapshot_id = 1;
   record.log_prefix = 10;

tests/fed/test_fed_replay.c

@@ -23,6 +23,8 @@ static amduat_fed_record_t make_record(uint32_t domain_id,
   record.meta.has_source = 0;
   record.id.type = type;
   record.id.ref = ref;
+  record.location.domain_id = domain_id;
+  record.location.ref = ref;
   record.logseq = logseq;
   record.snapshot_id = snapshot_id;
   record.log_prefix = log_prefix;

tests/fed/test_fed_view.c

@@ -65,6 +65,8 @@ static amduat_fed_record_t make_record(uint32_t domain_id,
   record.meta.has_source = 0;
   record.id.type = AMDUAT_FED_REC_ARTIFACT;
   record.id.ref = ref;
+  record.location.domain_id = domain_id;
+  record.location.ref = ref;
   record.logseq = logseq;
   record.snapshot_id = 1;
   record.log_prefix = 10;
@@ -124,7 +126,7 @@ static int test_view_and_resolve(void) {
   ops.get = stub_store_get;
   amduat_asl_store_init(&store, stub.config, ops, &stub);
 
-  if (amduat_fed_resolve(&view, &store, ref_a, &artifact) !=
+  if (amduat_fed_resolve(&view, &store, ref_a, &artifact, NULL) !=
       AMDUAT_FED_RESOLVE_OK) {
     fprintf(stderr, "resolve local failed\n");
     amduat_fed_view_free(&view);
@@ -132,14 +134,14 @@ static int test_view_and_resolve(void) {
   }
   amduat_artifact_free(&artifact);
 
-  if (amduat_fed_resolve(&view, &store, ref_c, &artifact) !=
+  if (amduat_fed_resolve(&view, &store, ref_c, &artifact, NULL) !=
       AMDUAT_FED_RESOLVE_FOUND_REMOTE_NO_BYTES) {
     fprintf(stderr, "resolve remote mismatch\n");
     amduat_fed_view_free(&view);
     return 1;
   }
 
-  if (amduat_fed_resolve(&view, &store, ref_d, &artifact) !=
+  if (amduat_fed_resolve(&view, &store, ref_d, &artifact, NULL) !=
       AMDUAT_FED_RESOLVE_POLICY_DENIED) {
     fprintf(stderr, "resolve policy denied mismatch\n");
     amduat_fed_view_free(&view);