Add Copy on Write (CoW) and use in event queue

codegenerator/cli/npm/envio/src/Cow.res

@@ -0,0 +1,45 @@
+type t<'a> = {
+  mutable data: 'a,
+  mutable isImmutable: bool,
+  copy: 'a => 'a,
+}
+
+let make = (data: 'a, ~copy): t<'a> => {data, isImmutable: true, copy}
+let getDataRef = (self: t<'a>): 'a => self.data
+let getData = (self: t<'a>): 'a =>
+  if self.isImmutable {
+    self.data
+  } else {
+    self.data->self.copy
+  }
+
+let copy = ({isImmutable, data, copy} as self: t<'a>): t<'a> =>
+  if isImmutable {
+    data->make(~copy)
+  } else {
+    self.isImmutable = true
+    data->make(~copy)
+  }
+
+let mutate = (self: t<'a>, fn) =>
+  if self.isImmutable {
+    self.isImmutable = false
+    self.data = self.data->self.copy
+    fn(self.data)
+  } else {
+    fn(self.data)
+  }
+
+module Array = {
+  type t<'a> = t<array<'a>>
+  module InternalFunctions = {
+    let copy = arr => arr->Array.copy
+    let push = (arr, item) => arr->Js.Array2.push(item)->ignore
+    let pop = arr => arr->Js.Array2.pop
+  }
+  let make = (data: array<'a>): t<'a> => make(data, ~copy=InternalFunctions.copy)
+  let push = (self: t<'a>, item) => self->mutate(arr => arr->InternalFunctions.push(item))
+  let pop = (self: t<'a>) => self->mutate(InternalFunctions.pop)
+  let length = (self: t<'a>) => self->getDataRef->Array.length
+  let last = (self: t<'a>) => self->getDataRef->Utils.Array.last
+}

codegenerator/cli/templates/static/codegen/src/eventFetching/FetchState.res

@@ -42,7 +42,7 @@ type partition = {
   //of a rollback.
   dynamicContracts: array<TablesStatic.DynamicContractRegistry.t>,
   //Events ordered from latest to earliest
-  fetchedEventQueue: array<Internal.eventItem>,
+  fetchedEventQueue: Cow.Array.t<Internal.eventItem>,
 }
 
 type t = {
@@ -62,7 +62,7 @@ type t = {
 
 let shallowCopyPartition = (p: partition) => {
   ...p,
-  fetchedEventQueue: p.fetchedEventQueue->Array.copy,
+  fetchedEventQueue: p.fetchedEventQueue->Cow.copy,
 }
 
 let copy = (fetchState: t) => {
@@ -155,7 +155,7 @@ let mergeIntoPartition = (p: partition, ~target: partition, ~maxAddrInPartition)
           status: {
             fetchingStateId: None,
           },
-          fetchedEventQueue: [],
+          fetchedEventQueue: []->Cow.Array.make,
           selection: target.selection,
           contractAddressMapping: restContractAddressMapping,
           dynamicContracts: restDcs,
@@ -178,7 +178,10 @@ let mergeIntoPartition = (p: partition, ~target: partition, ~maxAddrInPartition)
           selection: target.selection,
           contractAddressMapping: mergedContractAddressMapping,
           dynamicContracts: mergedDynamicContracts,
-          fetchedEventQueue: mergeSortedEventList(p.fetchedEventQueue, target.fetchedEventQueue),
+          fetchedEventQueue: mergeSortedEventList(
+            p.fetchedEventQueue->Cow.getDataRef,
+            target.fetchedEventQueue->Cow.getDataRef,
+          )->Cow.Array.make,
           latestFetchedBlock,
         },
         rest,
@@ -205,7 +208,10 @@ let addItemsToPartition = (
       fetchingStateId: None,
     },
     latestFetchedBlock,
-    fetchedEventQueue: Array.concat(reversedNewItems, p.fetchedEventQueue),
+    fetchedEventQueue: Array.concat(
+      reversedNewItems,
+      p.fetchedEventQueue->Cow.getDataRef,
+    )->Cow.Array.make,
   }
 }
 
@@ -255,7 +261,7 @@ let updateInternal = (
   for idx in 0 to partitions->Array.length - 1 {
     let p = partitions->Js.Array2.unsafe_get(idx)
 
-    let partitionQueueSize = p.fetchedEventQueue->Array.length
+    let partitionQueueSize = p.fetchedEventQueue->Cow.Array.length
 
     queueSize := queueSize.contents + partitionQueueSize
 
@@ -330,7 +336,7 @@ let makeDcPartition = (
     selection,
     contractAddressMapping,
     dynamicContracts,
-    fetchedEventQueue: [],
+    fetchedEventQueue: Cow.Array.make([]),
   }
 }
 
@@ -627,7 +633,7 @@ let getNextQuery = (
       if (
         p->checkIsFetchingPartition->not &&
         p.latestFetchedBlock.blockNumber < currentBlockHeight &&
-        (checkQueueSize ? p.fetchedEventQueue->Array.length < maxPartitionQueueSize : true) && (
+        (checkQueueSize ? p.fetchedEventQueue->Cow.Array.length < maxPartitionQueueSize : true) && (
           isWithinSyncRange
             ? true
             : !checkIsWithinSyncRange(~latestFetchedBlock=p.latestFetchedBlock, ~currentBlockHeight)
@@ -749,9 +755,9 @@ Returns queue item WITHOUT the updated fetch state. Used for checking values
 not updating state
 */
 let getEarliestEventInPartition = (p: partition) => {
-  switch p.fetchedEventQueue->Utils.Array.last {
+  switch p.fetchedEventQueue->Cow.Array.last {
   | Some(head) =>
-    Item({item: head, popItemOffQueue: () => p.fetchedEventQueue->Js.Array2.pop->ignore})
+    Item({item: head, popItemOffQueue: () => p.fetchedEventQueue->Cow.Array.pop->ignore})
   | None => makeNoItem(p)
   }
 }
@@ -819,7 +825,7 @@ let make = (
       },
       contractAddressMapping: ContractAddressingMap.make(),
       dynamicContracts: [],
-      fetchedEventQueue: [],
+      fetchedEventQueue: Cow.Array.make([]),
     })
   }
 
@@ -841,7 +847,7 @@ let make = (
           selection: normalSelection,
           contractAddressMapping: ContractAddressingMap.make(),
           dynamicContracts: [],
-          fetchedEventQueue: [],
+          fetchedEventQueue: Cow.Array.make([]),
         }
       }
 
@@ -980,12 +986,15 @@ let checkContainsRegisteredContractAddress = (
 let getLatestFullyFetchedBlock = ({latestFullyFetchedBlock}: t) => latestFullyFetchedBlock
 
 let pruneQueueFromFirstChangeEvent = (
-  queue: array<Internal.eventItem>,
+  queue: Cow.Array.t<Internal.eventItem>,
   ~firstChangeEvent: blockNumberAndLogIndex,
 ) => {
-  queue->Array.keep(item =>
+  queue
+  ->Cow.getDataRef
+  ->Array.keep(item =>
     (item.blockNumber, item.logIndex) < (firstChangeEvent.blockNumber, firstChangeEvent.logIndex)
   )
+  ->Cow.Array.make
 }
 
 /**

scenarios/test_codegen/test/ChainManager_test.res

@@ -370,7 +370,10 @@ describe("determineNextEvent", () => {
         selection: normalSelection,
         contractAddressMapping: ContractAddressingMap.make(),
         dynamicContracts: [],
-        fetchedEventQueue: item->Option.mapWithDefault([], v => [v]),
+        fetchedEventQueue: item->Option.mapWithDefault(
+          Cow.Array.make([]),
+          v => Cow.Array.make([v]),
+        ),
       }
       {
         partitions: [partition],

scenarios/test_codegen/test/lib_tests/Cow_test.res

@@ -0,0 +1,50 @@
+open RescriptMocha
+
+describe("Copy on write", () => {
+  it("Should not copy without mutation", () => {
+    let arr = [1, 2, 3]
+    let cowA = arr->Cow.Array.make
+    let cowB = cowA->Cow.copy
+
+    Assert.ok(
+      cowA->Cow.getDataRef === cowB->Cow.getDataRef && cowA->Cow.getDataRef === arr,
+      ~message="Should be a reference to the same array",
+    )
+
+    cowB->Cow.Array.push(4)
+
+    let cowBRefA = cowB->Cow.getDataRef
+
+    Assert.ok(
+      cowA->Cow.getDataRef === arr && arr !== cowB->Cow.getDataRef,
+      ~message="Should be a new array after mutation",
+    )
+
+    cowB->Cow.Array.push(5)
+    Assert.ok(
+      cowB->Cow.getDataRef === cowBRefA,
+      ~message="Should continue to be the same ref after mutation",
+    )
+
+    let cowC = cowB->Cow.copy
+
+    Assert.ok(
+      cowB->Cow.getDataRef === cowC->Cow.getDataRef,
+      ~message="Should be the same ref after copy",
+    )
+
+    cowB->Cow.Array.push(6)
+    Assert.ok(
+      cowB->Cow.getDataRef !== cowC->Cow.getDataRef,
+      ~message="Should not be the same ref after mutation",
+    )
+    Assert.ok(
+      cowC->Cow.getDataRef === cowBRefA,
+      ~message="See still has the original ref when it was copied",
+    )
+
+    Assert.deepEqual(cowA->Cow.getData, arr)
+    Assert.deepEqual(cowB->Cow.getData, [1, 2, 3, 4, 5, 6])
+    Assert.deepEqual(cowC->Cow.getData, [1, 2, 3, 4, 5])
+  })
+})