Timestamp Oracle (#9)

* init timestamp oracle

* Update Readme.md

* update stats, tests

* update aqua script and readme

ts-oracle/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "ts_oracle"
+version = "0.1.0"
+authors = ["boneyard93501 <4523011+boneyard93501@users.noreply.github.com>"]
+edition = "2018"
+description = "ts-cons, a Marine wasi module"
+license = "Apache-2.0"
+
+[[bin]]
+name = "ts_oracle"
+path = "src/main.rs"
+
+[dependencies]
+fluence = { version="0.6.9", features=["logger"]}
+log = "0.4.14"
+
+
+[dev-dependencies]
+fluence-test = "0.1.9"
+
+[dev]
+[profile.release]
+opt-level = "s"

ts-oracle/Config.toml

@@ -0,0 +1,6 @@
+modules_dir = "artifacts/"
+
+[[module]]
+    name = "ts_oracle"
+    mem_pages_count = 1
+    logger_enabled = true

ts-oracle/Readme.md

@@ -0,0 +1,170 @@
+# Timestamp Oracle PoC
+
+## Overview
+
+Getting accurate timestamps can be problematic in various contexts including blockchains. Timestamp oracles can alleviate this concern by aggregating and processing a variety of timestamp sources into a point- or range-estimate. Of course, the acquisition of accurate timestamps not subject to manipulation is a critical attribute of a good oracle.  The Fluence peer-to-peer network offers a large number of independent nodes that can serve as timestamp sources from either Kademilia or TrustGraph neighborhoods.
+
+**Note: The timestamps are currently NOT signed by the nodes. But that could be easily addressed.**
+
+### Fluence Solution
+Fluence Labs provides an open Web3 protocol, framework and associated tooling to develop and host applications, interfaces and backends on permissionless peer-to-peer networks. An integral part of the Fluence solution is the Aquamarine stack comprised of Aqua and Marine. 
+
+Aqua is a new programming language and paradigm purpose-built to program distributed networks and compose applications from distributed services. For more information on Aqua, see
+* [Aqua Book](https://app.gitbook.com/@fluence/s/aqua-book/)
+* [Aqua Playground](https://github.com/fluencelabs/aqua-playground)
+* [Aqua repo](https://github.com/fluencelabs/aqua)
+
+
+Marine is a general purpose Wasm runtime and toolkit, allows developers to build distributed services that can be composed into applications by Aqua. For more information on Marine, see
+* [Marine repo](https://github.com/fluencelabs/marine)
+* [Marine SDK](https://github.com/fluencelabs/marine-rs-sdk)
+
+### Setup
+
+*Please note that we already deployed the Rust service to node `12D3KooWHLxVhUQyAuZe6AHMB29P7wkvTNMn7eDMcsqimJYLKREf` with service id `ed657e45-0fe3-4d6c-b3a4-a2981b7cadb9`, which is all that's needed to use the service in Aqua.*
+
+In order to run the entire code base, Rust and Node required. If necessary see [Install Rust](https://www.rust-lang.org/tools/install) and [NVM](https://github.com/nvm-sh/nvm) for details.
+
+Install the following tools:
+
+```bash
+cargo install mrepl 
+cargo install marine
+
+npm -g install @fluencelabs/aqua-cli
+npm -g install @fluencelabs/flidst
+```
+
+To compile the code to the wasi target:
+
+```bash
+./scripts/build.sh
+```
+
+You can use the REPL to locally test the services:
+
+```bash
+mrepl Config.toml
+```
+
+test the code with
+
+```bash
+cargo +nightly test --release
+```
+
+and deploy the service to a peer of your choice with the `fldist` tool:
+
+```bash
+fldist new_service --ms artifacts/ts_oracle.wasm:artifacts/ts_oracle_cfg.json --name ts-consensus --verbose
+```
+
+and to compile the Aqua scripts:
+
+```bash
+# results in air-scripts ts_getter.ts
+aqua-cli -i aqua-scripts -o air-scripts 
+```
+
+which generates Typescript code wrapping the compiled Aqua intermediary representation (AIR) or 
+
+```bash
+# results in air-scripts 
+# ts_getter.ts_getter.air and ts_getter.ts_oracle.air
+aqua-cli -i aqua-scripts -o air-scripts -a
+```
+
+which generates raw AIR files.
+
+### Approach
+
+We implemented a custom service that returns the mode and frequency for an array of timestamps, see `src/` that can be deployed on to any node of the peer-to-peer network and, once deployed, used to in an Aqua script. Moreover, network peers have builtin services including Kademlia and timestamp services. Both custom and bultin services are accessible by Aqua and ready for composition into an application.
+
+Our oracle solution is implemented in Aqua and utilizes timestamps from peers selected from our Kademlia neighborhood and, for illustrative purposes, use the deployed service to arrive at the point estimate for our oracle. See `src/main.rs`. There certanly are better ways to process the timestamps into an oracle but for our purposes, mode works.
+
+In our Aqua script, `aqua-scripts/ts_getter`, we separate the timestamp collections from the subsequent oracle processing. That is, if a peer-client wants to process the timestamps locally, all that's needed are the timestamps, which can be obtained by calling the `ts_getter` function. Alternatively, the timestamps may be processed by calling one or more `ts-oracle` services deployed to the network.
+
+```aqua
+-- aqua-scripts/ts_getter.aqua
+import "builtin.aqua"
+
+service Op2("op"):
+    identity(s: u64)
+     array(a: string, b: u64) -> string
+
+-- the data struct from the Wasm file
+-- marine aqua artifacts/ts_oracle.wasm
+data Oracle:
+    n: u32
+    mode: u64
+    freq: u32
+    err_str: string
+    raw_data: []u64
+
+-- the point_estimate function from the Wasm file wrapped as a service
+-- marine aqua artifacts/ts_oracle.wasm
+service TSOracle("service-id"):
+    point_estimate: []u64, u32 -> Oracle
+
+-- the timestamp getter drawing from the Kademlia neighborhood
+-- function collects the neighborhood peers
+-- retrieves and returns the timestamps 
+func ts_getter(node: string) -> []u64:
+  res: *u64
+  on node:
+    k <- Op.string_to_b58(node)
+    nodes <- Kademlia.neighborhood(k, false)
+    for n <- nodes par:
+      on n:
+        try:
+          res <- Peer.timestamp_ms()
+    Op2.identity(res!9)
+  <- res
+
+-- oracle function operating on array of timestamps utilizing a distributed
+-- service to calculate the point_estimate
+-- see src/main.rs for the service implementation
+func ts_oracle(node: string, oracle_service_id: string, min_points:u32) -> Oracle:
+  res <- ts_getter(node)
+  
+  on node:
+    TSOracle oracle_service_id
+    oracle <- TSOracle.point_estimate(res, min_points)  -- calculate mode 
+  <- oracle                                             -- and return to initiating peer
+```
+
+We can run our Aqua `ts_oracle` script against the deployed processing service to get our oracle point estimate:
+
+```bash
+fldist run_air -p air-scripts/ts_getter.ts_oracle.air  -d '{"node":"12D3KooWHLxVhUQyAuZe6AHMB29P7wkvTNMn7eDMcsqimJYLKREf", "oracle_service_id":"ed657e45-0fe3-4d6c-b3a4-a2981b7cadb9", "min_points":5}' --generated
+[
+  {
+    "err_str": "",
+    "freq": 2,              -- this may change
+    "mode": 1623713287898,  -- this changes, of course
+    "n": 10
+  }
+]
+```
+
+or run the `ts_getter` functions just for the timestamps:
+
+```aqua
+fldist run_air -p air-scripts/ts_getter.ts_getter.air  -d '{"node":"12D3KooWHLxVhUQyAuZe6AHMB29P7wkvTNMn7eDMcsqimJYLKREf", "oracle_service_id":"ed657e45-0fe3-4d6c-b3a4-a2981b7cadb9", "min_points":5, "n_ts": 10}' --generated
+[
+  [
+    1624834792801,
+    1624834792791,
+    1624834792796,
+    1624834792797,
+    1624834792795,
+    1624834792783,
+    1624834792800,
+    1624834792785,
+    1624834792806,
+    1624834792793
+  ]
+]
+```
+
+Instead of the `fldist`  cli, you can use the Typescript stub and integrate it into a TS client. See [Aqua Playground](https://github.com/fluencelabs/aqua-playground) for more information.

ts-oracle/air-scripts/ts_getter.ts

@@ -0,0 +1,206 @@
+/**
+ *
+ * This file is auto-generated. Do not edit manually: changes may be erased.
+ * Generated by Aqua compiler: https://github.com/fluencelabs/aqua/. 
+ * If you find any bugs, please write an issue on GitHub: https://github.com/fluencelabs/aqua/issues
+ * Aqua version: 0.1.8-157
+ *
+ */
+import { FluenceClient, PeerIdB58 } from '@fluencelabs/fluence';
+import { RequestFlowBuilder } from '@fluencelabs/fluence/dist/api.unstable';
+import { RequestFlow } from '@fluencelabs/fluence/dist/internal/RequestFlow';
+
+
+
+export async function ts_getter(client: FluenceClient, node: string, config?: {ttl?: number}): Promise<number[]> {
+    let request: RequestFlow;
+    const promise = new Promise<number[]>((resolve, reject) => {
+        request = new RequestFlowBuilder()
+            .disableInjections()
+            .withTTL(config?.ttl || 5000)
+            .withRawScript(
+                `
+(xor
+ (seq
+  (seq
+   (seq
+    (seq
+     (seq
+      (call %init_peer_id% ("getDataSrv" "-relay-") [] -relay-)
+      (call %init_peer_id% ("getDataSrv" "node") [] node)
+     )
+     (call -relay- ("op" "noop") [])
+    )
+    (xor
+     (seq
+      (seq
+       (seq
+        (call node ("op" "string_to_b58") [node] k)
+        (call node ("kad" "neighborhood") [k false] nodes)
+       )
+       (fold nodes n
+        (par
+         (seq
+          (xor
+           (call n ("peer" "timestamp_ms") [] $res)
+           (null)
+          )
+          (call node ("op" "noop") [])
+         )
+         (next n)
+        )
+       )
+      )
+      (call node ("op" "identity") [$res.$.[9]!])
+     )
+     (seq
+      (call -relay- ("op" "noop") [])
+      (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 1])
+     )
+    )
+   )
+   (call -relay- ("op" "noop") [])
+  )
+  (xor
+   (call %init_peer_id% ("callbackSrv" "response") [$res])
+   (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 2])
+  )
+ )
+ (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 3])
+)
+
+            `,
+            )
+            .configHandler((h) => {
+                h.on('getDataSrv', '-relay-', () => {
+                    return client.relayPeerId!;
+                });
+                h.on('getDataSrv', 'node', () => {return node;});
+                h.onEvent('callbackSrv', 'response', (args) => {
+  const [res] = args;
+  resolve(res);
+});
+
+                h.onEvent('errorHandlingSrv', 'error', (args) => {
+                    // assuming error is the single argument
+                    const [err] = args;
+                    reject(err);
+                });
+            })
+            .handleScriptError(reject)
+            .handleTimeout(() => {
+                reject('Request timed out for ts_getter');
+            })
+            .build();
+    });
+    await client.initiateFlow(request!);
+    return promise;
+}
+      
+
+
+export async function ts_oracle(client: FluenceClient, node: string, oracle_service_id: string, min_points: number, config?: {ttl?: number}): Promise<{err_str:string;freq:number;mode:number;n:number;raw_data:number[]}> {
+    let request: RequestFlow;
+    const promise = new Promise<{err_str:string;freq:number;mode:number;n:number;raw_data:number[]}>((resolve, reject) => {
+        request = new RequestFlowBuilder()
+            .disableInjections()
+            .withTTL(config?.ttl || 5000)
+            .withRawScript(
+                `
+(xor
+ (seq
+  (seq
+   (seq
+    (seq
+     (seq
+      (seq
+       (seq
+        (seq
+         (call %init_peer_id% ("getDataSrv" "-relay-") [] -relay-)
+         (call %init_peer_id% ("getDataSrv" "node") [] node)
+        )
+        (call %init_peer_id% ("getDataSrv" "oracle_service_id") [] oracle_service_id)
+       )
+       (call %init_peer_id% ("getDataSrv" "min_points") [] min_points)
+      )
+      (call -relay- ("op" "noop") [])
+     )
+     (xor
+      (seq
+       (seq
+        (seq
+         (call node ("op" "string_to_b58") [node] k)
+         (call node ("kad" "neighborhood") [k false] nodes)
+        )
+        (fold nodes n
+         (par
+          (seq
+           (xor
+            (call n ("peer" "timestamp_ms") [] $res)
+            (null)
+           )
+           (call node ("op" "noop") [])
+          )
+          (next n)
+         )
+        )
+       )
+       (call node ("op" "identity") [$res.$.[9]!])
+      )
+      (seq
+       (seq
+        (call -relay- ("op" "noop") [])
+        (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 1])
+       )
+       (call -relay- ("op" "noop") [])
+      )
+     )
+    )
+    (xor
+     (call node (oracle_service_id "point_estimate") [$res min_points] oracle)
+     (seq
+      (call -relay- ("op" "noop") [])
+      (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 2])
+     )
+    )
+   )
+   (call -relay- ("op" "noop") [])
+  )
+  (xor
+   (call %init_peer_id% ("callbackSrv" "response") [oracle])
+   (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 3])
+  )
+ )
+ (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 4])
+)
+
+            `,
+            )
+            .configHandler((h) => {
+                h.on('getDataSrv', '-relay-', () => {
+                    return client.relayPeerId!;
+                });
+                h.on('getDataSrv', 'node', () => {return node;});
+h.on('getDataSrv', 'oracle_service_id', () => {return oracle_service_id;});
+h.on('getDataSrv', 'min_points', () => {return min_points;});
+                h.onEvent('callbackSrv', 'response', (args) => {
+  const [res] = args;
+  resolve(res);
+});
+
+                h.onEvent('errorHandlingSrv', 'error', (args) => {
+                    // assuming error is the single argument
+                    const [err] = args;
+                    reject(err);
+                });
+            })
+            .handleScriptError(reject)
+            .handleTimeout(() => {
+                reject('Request timed out for ts_oracle');
+            })
+            .build();
+    });
+    await client.initiateFlow(request!);
+    return promise;
+}
+      

ts-oracle/air-scripts/ts_getter.ts_getter.air

@@ -0,0 +1,48 @@
+(xor
+ (seq
+  (seq
+   (seq
+    (seq
+     (seq
+      (call %init_peer_id% ("getDataSrv" "-relay-") [] -relay-)
+      (call %init_peer_id% ("getDataSrv" "node") [] node)
+     )
+     (call -relay- ("op" "noop") [])
+    )
+    (xor
+     (seq
+      (seq
+       (seq
+        (call node ("op" "string_to_b58") [node] k)
+        (call node ("kad" "neighborhood") [k false] nodes)
+       )
+       (fold nodes n
+        (par
+         (seq
+          (xor
+           (call n ("peer" "timestamp_ms") [] $res)
+           (null)
+          )
+          (call node ("op" "noop") [])
+         )
+         (next n)
+        )
+       )
+      )
+      (call node ("op" "identity") [$res.$.[9]!])
+     )
+     (seq
+      (call -relay- ("op" "noop") [])
+      (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 1])
+     )
+    )
+   )
+   (call -relay- ("op" "noop") [])
+  )
+  (xor
+   (call %init_peer_id% ("callbackSrv" "response") [$res])
+   (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 2])
+  )
+ )
+ (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 3])
+)

ts-oracle/air-scripts/ts_getter.ts_oracle.air

@@ -0,0 +1,66 @@
+(xor
+ (seq
+  (seq
+   (seq
+    (seq
+     (seq
+      (seq
+       (seq
+        (seq
+         (call %init_peer_id% ("getDataSrv" "-relay-") [] -relay-)
+         (call %init_peer_id% ("getDataSrv" "node") [] node)
+        )
+        (call %init_peer_id% ("getDataSrv" "oracle_service_id") [] oracle_service_id)
+       )
+       (call %init_peer_id% ("getDataSrv" "min_points") [] min_points)
+      )
+      (call -relay- ("op" "noop") [])
+     )
+     (xor
+      (seq
+       (seq
+        (seq
+         (call node ("op" "string_to_b58") [node] k)
+         (call node ("kad" "neighborhood") [k false] nodes)
+        )
+        (fold nodes n
+         (par
+          (seq
+           (xor
+            (call n ("peer" "timestamp_ms") [] $res)
+            (null)
+           )
+           (call node ("op" "noop") [])
+          )
+          (next n)
+         )
+        )
+       )
+       (call node ("op" "identity") [$res.$.[9]!])
+      )
+      (seq
+       (seq
+        (call -relay- ("op" "noop") [])
+        (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 1])
+       )
+       (call -relay- ("op" "noop") [])
+      )
+     )
+    )
+    (xor
+     (call node (oracle_service_id "point_estimate") [$res min_points] oracle)
+     (seq
+      (call -relay- ("op" "noop") [])
+      (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 2])
+     )
+    )
+   )
+   (call -relay- ("op" "noop") [])
+  )
+  (xor
+   (call %init_peer_id% ("callbackSrv" "response") [oracle])
+   (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 3])
+  )
+ )
+ (call %init_peer_id% ("errorHandlingSrv" "error") [%last_error% 4])
+)

ts-oracle/aqua-scripts/builtin.aqua

@@ -0,0 +1,182 @@
+-- Default public interface of Fluence nodes
+
+alias Field : []string
+alias Argument : []string
+alias Bytes : []u8
+alias PeerId : string
+
+data Service:
+    id: string
+    blueprint_id: string
+    owner_id: string
+
+data FunctionSignature:
+    arguments: []Argument
+    name: string
+    output_types: []string
+
+data RecordType:
+    fields: []Field
+    id: u64
+    name: string
+
+data Interface:
+    function_signatures: []FunctionSignature
+    record_types: []RecordType
+
+data ServiceInfo:
+    blueprint_id: string
+    service_id: string
+    interface: Interface
+
+data Info:
+    external_addresses: []string
+
+data ModuleConfig:
+    name: string
+
+data Module:
+    name: string
+    hash: string
+    config: ModuleConfig
+
+data AddBlueprint:
+    name: string
+    dependencies: []string
+
+data Blueprint:
+    id: string
+    name: string
+    dependencies: []string
+
+data ScriptInfo:
+    id: string
+    src: string
+    failures: u32
+    interval: string
+    owner: string
+
+data Contact:
+    peer_id: string
+    addresses: []string
+
+service Op("op"):
+    -- does nothing
+    noop()
+    -- takes any number of arguments and wraps them into a single array
+    array(a: string, b: string, c: string) -> []string
+    -- takes any number of arrays and flattens them by concatenating
+    concat(a: []string, b: []string, c: []string) -> []string
+    -- takes a single argument and returns it back
+    identity(s: ?string) -> ?string
+    string_to_b58(s: string) -> string
+    string_from_b58(b: string) -> string
+    bytes_to_b58(bs: []u8) -> string
+    bytes_from_b58(b: string) -> []u8
+    -- Applies SHA256 to the given string
+    -- Argument: s - string to apply sha256 to
+    -- Returns: returns sha256 multihash encoded as base58
+    sha256_string(s: string) -> string
+
+service Peer("peer"):
+    -- Checks if there is a direct connection to the peer identified by a given PeerId
+    -- Argument: PeerId – id of the peer to check if there's a connection with
+    -- Returns: bool - true if connected to the peer, false otherwise
+    is_connected(peer: PeerId) -> bool
+
+    -- Initiates a connection to the specified peer
+    -- Arguments:
+    -- PeerId – id of the target peer
+    -- [Multiaddr] – an array of target peer's addresses
+    -- Returns: bool - true if connection was successful
+    connect(id: PeerId, multiaddrs: []string) -> bool
+    -- Resolves the contact of a peer via Kademlia
+    -- Argument: PeerId – id of the target peer
+    -- Returns: Contact - true if connection was successful
+    get_contact(peer: PeerId) -> Contact
+
+    -- Get information about the peer
+    identify() -> Info
+
+    -- Get Unix timestamp in milliseconds
+    timestamp_ms() -> u64
+
+    -- Get Unix timestamp in seconds
+    timestamp_sec() -> u64
+
+service Kademlia("kad"):
+    -- Instructs node to return the locally-known nodes
+    -- in the Kademlia neighborhood for a given key
+    neighborhood(key: PeerId, already_hashed: bool) -> []PeerId
+    -- Merges given lists and sorts them by distance to target
+    -- Arguments:
+    -- target – base58 string; result is sorted by XOR distance to target
+    -- left – list of base58 strings
+    -- right – list of base58 strings
+    -- count – how many items to return
+    -- Returns: list of base58 strings sorted by distance to target; list will contain at most count elements
+    merge(target: string, left: []string, right: []string, count: u32) -> []string
+
+service Srv("srv"):
+    -- Used to create a service on a certain node
+    -- Arguments:
+    -- blueprint_id – ID of the blueprint that has been added to the node specified in the service call by the dist add_blueprint service.
+    -- Returns: service_id – the service ID of the created service.
+    create(blueprint_id: string) -> string
+
+    -- Used to remove a service from a certain node
+    -- Arguments:
+    -- service_id – ID of the service to remove
+    remove(service_id: string)
+
+    -- Returns a list of services running on a peer
+    list() -> []Service
+
+    -- Adds an alias on service, so, service could be called
+    -- not only by service_id but by alias as well.
+    -- Argument:
+    -- alias - settable service name
+    -- service_id – ID of the service whose interface you want to name.
+    add_alias(alias: string, service_id: string)
+
+    -- Resolves given alias to a service id
+    -- If there's no such alias, throws an error
+    -- Returns: service id associated with the given alias
+    resolve_alias(alias: string) -> string
+
+    -- Retrieves the functional interface of a service running
+    -- on the node specified in the service call
+    -- Argument: service_id – ID of the service whose interface you want to retrieve.
+    get_interface(service_id: string) -> ServiceInfo
+
+service Dist("dist"):
+    -- Used to add modules to the node specified in the service call
+    -- Arguments:
+    -- bytes – a base64 string containing the .wasm module to add.
+    -- config – module info
+    -- Returns: blake3 hash of the module
+    add_module(wasm_b56_content: Bytes, conf: ModuleConfig) -> string
+
+    -- Get a list of modules available on the node
+    list_modules() -> []Module
+
+    -- Get the interface of a module
+    get_interface(module_id: string) -> Interface
+
+    -- Used to add a blueprint to the node specified in the service call
+    add_blueprint(blueprint: AddBlueprint) -> string
+
+    -- Used to get the blueprints available on the node specified in the service call.
+    -- A blueprint is an object of the following structure
+    list_blueprints() -> []Blueprint
+
+service Script("script"):
+    -- Adds the given script to a node
+    add(air_content: string, run_every: string) -> string
+
+    -- Removes recurring script from a node. Only a creator of the script can delete it
+    remove(script_id: string) -> bool
+
+    -- Returns a list of existing scripts on the node.
+    -- Each object in the list is of the following structure
+    list() -> ScriptInfo

ts-oracle/aqua-scripts/ts_getter.aqua

@@ -0,0 +1,37 @@
+import "builtin.aqua"
+
+service Op2("op"):
+    identity(s: u64)
+     array(a: string, b: u64) -> string
+
+data Oracle:
+    n: u32
+    mode: u64
+    freq: u32
+    err_str: string
+    raw_data: []u64
+
+service TSOracle("service-id"):
+    point_estimate: []u64, u32 -> Oracle
+
+
+func ts_getter(node: string) -> []u64:
+  res: *u64
+  on node:
+    k <- Op.string_to_b58(node)
+    nodes <- Kademlia.neighborhood(k, false)
+    for n <- nodes par:
+      on n:
+        try:
+          res <- Peer.timestamp_ms()
+    Op2.identity(res!9)
+  <- res
+
+func ts_oracle(node: string, oracle_service_id: string, min_points:u32) -> Oracle:
+  res <- ts_getter(node)
+  
+  on node:
+    TSOracle oracle_service_id
+    oracle <- TSOracle.point_estimate(res, min_points)
+  <- oracle
+

ts-oracle/artifacts/service_info.txt

@@ -0,0 +1,6 @@
+fldist new_service --ms artifacts/ts_oracle.wasm:artifacts/ts_oracle_cfg.json --name ts-consensus --verbose
+client seed: 7EbFfPnbjZHBWmkDT5dc5mXfY4ntFxiQ9hwynYxjL1S4
+client peerId: 12D3KooWAw1obs4xApwechRaz7GEqTA54JLTq8KR7MJwRShjxBaH
+relay peerId: 12D3KooWHLxVhUQyAuZe6AHMB29P7wkvTNMn7eDMcsqimJYLKREf
+service id: ed657e45-0fe3-4d6c-b3a4-a2981b7cadb9
+service created successfully

ts-oracle/artifacts/ts_oracle_cfg.json

@@ -0,0 +1,3 @@
+{
+    "name": "ts_oracle"
+}

ts-oracle/scripts/build.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash -o errexit -o nounset -o pipefail
+
+mkdir -p artifacts
+rm -f artifacts/*.wasm
+marine build --release
+cp target/wasm32-wasi/release/ts_oracle.wasm artifacts/

ts-oracle/src/main.rs

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2021 Fluence Labs Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+use fluence::{marine, module_manifest, WasmLoggerBuilder};
+
+mod stats;
+
+module_manifest!();
+
+pub fn main() {
+    WasmLoggerBuilder::new().build().unwrap();
+}
+
+#[marine]
+#[derive(Default, Debug)]
+pub struct Oracle {
+    pub n: u32,
+    pub mode: u64,
+    pub freq: u32,
+    pub err_str: String,
+}
+
+#[marine]
+pub fn point_estimate(tstamps: Vec<u64>, min_points: u32) -> Oracle {
+    if tstamps.len() < min_points as usize {
+        return Oracle {
+            err_str: format!(
+                "Expected at least {} points but only got {}.",
+                min_points,
+                tstamps.len()
+            ),
+            ..<_>::default()
+        };
+    }
+
+    if tstamps.len() < 1 {
+        return Oracle {
+            err_str: format!("Expected at least one timestamp."),
+            ..<_>::default()
+        };
+    }
+
+    let (freq, mode) = stats::mode(tstamps.iter());
+
+    Oracle {
+        n: tstamps.len() as u32,
+        mode,
+        freq,
+
+        ..<_>::default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use fluence_test::marine_test;
+
+    #[test]
+    fn test_mean_good() {
+        let data = vec![1u64, 2u64, 3u64];
+        let res = stats::mean(data.iter());
+        assert!(res.is_some());
+        assert_eq!(res.unwrap(), 2f64)
+    }
+
+    #[test]
+    fn test_mean_bad() {
+        let data = vec![];
+        let res = stats::mean(data.iter());
+        assert!(res.is_none());
+    }
+
+    #[test]
+    fn test_mode() {
+        let data = vec![1u64, 1u64, 3u64, 3u64, 3u64, 5u64];
+        let (freq, mode) = stats::mode(data.iter());
+        assert_eq!(mode, 3u64);
+        assert_eq!(freq, 3);
+    }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts")]
+    fn test_point_estimate_good() {
+        let data = vec![1u64, 1u64, 3u64, 3u64, 3u64, 5u64];
+        let min_points = 2u32;
+        let res = ts_oracle.point_estimate(data, min_points);
+        assert_eq!(res.mode, 3u64);
+        assert_eq!(res.freq, 3u32);
+    }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts")]
+    fn test_point_estimate_bad() {
+        let data = vec![1u64, 1u64, 3u64, 3u64, 3u64, 5u64];
+        let n = data.len();
+        let min_points = 20u32;
+        let res = ts_oracle.point_estimate(data, min_points);
+        assert_eq!(
+            res.err_str,
+            format!(
+                "Expected at least {} points but only got {}.",
+                min_points, n
+            )
+        );
+    }
+
+    #[marine_test(config_path = "../Config.toml", modules_dir = "../artifacts")]
+    fn test_point_estimate_bad_2() {
+        let data = vec![];
+        let n = data.len();
+        let min_points = 0u32;
+        let res = ts_oracle.point_estimate(data, min_points);
+        println!("res: {:?}", res);
+        assert_eq!(res.err_str, "Expected at least one timestamp.".to_string());
+    }
+}

ts-oracle/src/stats.rs

@@ -0,0 +1,28 @@
+use std::collections::HashMap;
+
+pub fn mode<'a>(data: impl ExactSizeIterator<Item = &'a u64>) -> (u32, u64) {
+    let frequencies = data
+        .into_iter()
+        .fold(HashMap::<u64, u32>::new(), |mut freqs, value| {
+            *freqs.entry(*value).or_insert(0) += 1;
+            freqs
+        });
+
+    let mode = frequencies
+        .clone()
+        .into_iter()
+        .max_by_key(|&(_, count)| count)
+        .map(|(value, _)| value)
+        .unwrap();
+
+    (*frequencies.get(&mode).unwrap(), mode)
+}
+
+pub fn mean<'a>(data: impl ExactSizeIterator<Item = &'a u64>) -> Option<f64> {
+    let n = data.len() as u64;
+    if n < 1 {
+        return None;
+    }
+    let res = (data.sum::<u64>() / n) as f64;
+    Some(res)
+}