Support Batch Price attestation for terra relay (#75)

* Support Batch Price attestation for terra relay

Tiltfile

@@ -258,7 +258,7 @@ if pyth:
     k8s_resource(
         "pyth", 
         resource_deps = ["solana-devnet"], 
-        labels = ["solana"],
+        labels = ["pyth"],
         trigger_mode = trigger_mode,
     )
 
@@ -283,7 +283,7 @@ if pyth:
         "p2w-attest",
         resource_deps = ["solana-devnet", "pyth", "guardian"],
         port_forwards = [],
-        labels = ["solana"],
+        labels = ["pyth"],
         trigger_mode = trigger_mode,
     )
 
@@ -292,8 +292,26 @@ if pyth:
         "p2w-relay",
         resource_deps = ["solana-devnet", "eth-devnet", "pyth", "guardian", "p2w-attest", "proto-gen-web", "wasm-gen"],
         port_forwards = [],
+        labels = ["pyth"]
     )
 
+    # Terra relay
+    docker_build(
+        ref = "p2w-terra-relay",
+        context = "third_party/pyth/p2w-terra-relay",
+        dockerfile = "third_party/pyth/p2w-terra-relay/Dockerfile.pyth_relay",
+    )
+    k8s_yaml_with_ns("devnet/p2w-terra-relay.yaml")
+    k8s_resource(
+        "p2w-terra-relay",
+        resource_deps = ["pyth", "p2w-attest", "spy", "terra-terrad"],
+        port_forwards = [
+            port_forward(4200, name = "Rest API (Status + Query) [:4200]", host = webHost),
+            port_forward(8081, name = "Prometheus [:8081]", host = webHost)],
+        labels = ["pyth"]
+    )
+
+
 k8s_yaml_with_ns("devnet/eth-devnet.yaml")
 
 k8s_resource(

devnet/p2w-terra-relay.yaml

@@ -0,0 +1,80 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: p2w-terra-relay
+  labels:
+    app: p2w-terra-relay
+spec:
+  ports:
+    - port: 8081
+      name: prometheus
+      protocol: TCP
+    - port: 4200
+      name: rest-api
+      protocol: TCP
+  clusterIP: None
+  selector:
+    app: p2w-terra-relay
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: p2w-terra-relay
+spec:
+  selector:
+    matchLabels:
+      app: p2w-terra-relay
+  serviceName: p2w-terra-relay
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: p2w-terra-relay
+    spec:
+      terminationGracePeriodSeconds: 0
+      containers:
+        - name: p2w-terra-relay
+          image: p2w-terra-relay
+          ports:
+            - containerPort: 8081
+              name: prometheus
+              protocol: TCP
+            - containerPort: 4200
+              name: rest-api
+              protocol: TCP
+          tcpSocket:
+            port: 2000
+          env:
+            - name: SPY_SERVICE_HOST
+              value: spy:7072
+            - name: SPY_SERVICE_FILTERS
+              value: '[{"chain_id":1,"emitter_address":"71f8dcb863d176e2c420ad6610cf687359612b6fb392e0642b0ca6b1f186aa3b"}]'
+            - name: TERRA_NODE_URL
+              value: http://terra-terrad:1317
+            - name: TERRA_PRIVATE_KEY
+              value: notice oak worry limit wrap speak medal online prefer cluster roof addict wrist behave treat actual wasp year salad speed social layer crew genius
+            - name: TERRA_PYTH_CONTRACT_ADDRESS
+              value: terra1plju286nnfj3z54wgcggd4enwaa9fgf5kgrgzl
+            - name: TERRA_CHAIN_ID
+              value: localterra
+            - name: TERRA_NAME
+              value: localterra
+            - name: TERRA_COIN
+              value: uluna
+            - name: REST_PORT
+              value: '4200'
+            - name: PROM_PORT
+              value: '8081'
+            - name: BAL_QUERY_INTERVAL
+              value: '60000'
+            - name: READINESS_PORT
+              value: '2000'
+            - name: RETRY_MAX_ATTEMPTS
+              value: '4'
+            - name: RETRY_DELAY_IN_MS
+              value: '250'
+            - name: MAX_MSGS_PER_BATCH
+              value: '1'
+            - name: LOG_LEVEL
+              value: debug

third_party/pyth/p2w-terra-relay/package-lock.json

@@ -14,6 +14,7 @@
         "@solana/spl-token": "^0.1.8",
         "@solana/web3.js": "^1.24.0",
         "@terra-money/terra.js": "^3.0.4",
+        "@types/express": "^4.17.13",
         "async-mutex": "^0.3.2",
         "body-parser": "^1.19.0",
         "condition-variable": "^1.0.0",
@@ -2273,6 +2274,15 @@
         "@types/node": "*"
       }
     },
+    "node_modules/@types/body-parser": {
+      "version": "1.19.2",
+      "resolved": "https://registry.npmjs.org/@types/body-parser/-/body-parser-1.19.2.tgz",
+      "integrity": "sha512-ALYone6pm6QmwZoAgeyNksccT9Q4AWZQ6PvfwR37GT6r6FWUPguq6sUmNGSMV2Wr761oQoBxwGGa6DR5o1DC9g==",
+      "dependencies": {
+        "@types/connect": "*",
+        "@types/node": "*"
+      }
+    },
     "node_modules/@types/connect": {
       "version": "3.4.35",
       "resolved": "https://registry.npmjs.org/@types/connect/-/connect-3.4.35.tgz",
@@ -2281,6 +2291,17 @@
         "@types/node": "*"
       }
     },
+    "node_modules/@types/express": {
+      "version": "4.17.13",
+      "resolved": "https://registry.npmjs.org/@types/express/-/express-4.17.13.tgz",
+      "integrity": "sha512-6bSZTPaTIACxn48l50SR+axgrqm6qXFIxrdAKaG6PaJk3+zuUr35hBlgT7vOmJcum+OEaIBLtHV/qloEAFITeA==",
+      "dependencies": {
+        "@types/body-parser": "*",
+        "@types/express-serve-static-core": "^4.17.18",
+        "@types/qs": "*",
+        "@types/serve-static": "*"
+      }
+    },
     "node_modules/@types/express-serve-static-core": {
       "version": "4.17.26",
       "resolved": "https://registry.npmjs.org/@types/express-serve-static-core/-/express-serve-static-core-4.17.26.tgz",
@@ -2344,6 +2365,11 @@
       "resolved": "https://registry.npmjs.org/@types/long/-/long-4.0.1.tgz",
       "integrity": "sha512-5tXH6Bx/kNGd3MgffdmP4dy2Z+G4eaXw0SE81Tq3BNadtnMR5/ySMzX4SLEzHJzSmPNn4HIdpQsBvXMUykr58w=="
     },
+    "node_modules/@types/mime": {
+      "version": "1.3.2",
+      "resolved": "https://registry.npmjs.org/@types/mime/-/mime-1.3.2.tgz",
+      "integrity": "sha512-YATxVxgRqNH6nHEIsvg6k2Boc1JHI9ZbH5iWFFv/MTkchz3b1ieGDa5T0a9RznNdI0KhVbdbWSN+KWWrQZRxTw=="
+    },
     "node_modules/@types/node": {
       "version": "16.11.11",
       "resolved": "https://registry.npmjs.org/@types/node/-/node-16.11.11.tgz",
@@ -2365,6 +2391,15 @@
       "resolved": "https://registry.npmjs.org/@types/range-parser/-/range-parser-1.2.4.tgz",
       "integrity": "sha512-EEhsLsD6UsDM1yFhAvy0Cjr6VwmpMWqFBCb9w07wVugF7w9nfajxLuVmngTIpgS6svCnm6Vaw+MZhoDCKnOfsw=="
     },
+    "node_modules/@types/serve-static": {
+      "version": "1.13.10",
+      "resolved": "https://registry.npmjs.org/@types/serve-static/-/serve-static-1.13.10.tgz",
+      "integrity": "sha512-nCkHGI4w7ZgAdNkrEu0bv+4xNV/XDqW+DydknebMOQwkpDGx8G+HTlj7R7ABI8i8nKxVw0wtKPi1D+lPOkh4YQ==",
+      "dependencies": {
+        "@types/mime": "^1",
+        "@types/node": "*"
+      }
+    },
     "node_modules/@types/stack-utils": {
       "version": "2.0.1",
       "resolved": "https://registry.npmjs.org/@types/stack-utils/-/stack-utils-2.0.1.tgz",
@@ -9260,6 +9295,15 @@
         "@types/node": "*"
       }
     },
+    "@types/body-parser": {
+      "version": "1.19.2",
+      "resolved": "https://registry.npmjs.org/@types/body-parser/-/body-parser-1.19.2.tgz",
+      "integrity": "sha512-ALYone6pm6QmwZoAgeyNksccT9Q4AWZQ6PvfwR37GT6r6FWUPguq6sUmNGSMV2Wr761oQoBxwGGa6DR5o1DC9g==",
+      "requires": {
+        "@types/connect": "*",
+        "@types/node": "*"
+      }
+    },
     "@types/connect": {
       "version": "3.4.35",
       "resolved": "https://registry.npmjs.org/@types/connect/-/connect-3.4.35.tgz",
@@ -9268,6 +9312,17 @@
         "@types/node": "*"
       }
     },
+    "@types/express": {
+      "version": "4.17.13",
+      "resolved": "https://registry.npmjs.org/@types/express/-/express-4.17.13.tgz",
+      "integrity": "sha512-6bSZTPaTIACxn48l50SR+axgrqm6qXFIxrdAKaG6PaJk3+zuUr35hBlgT7vOmJcum+OEaIBLtHV/qloEAFITeA==",
+      "requires": {
+        "@types/body-parser": "*",
+        "@types/express-serve-static-core": "^4.17.18",
+        "@types/qs": "*",
+        "@types/serve-static": "*"
+      }
+    },
     "@types/express-serve-static-core": {
       "version": "4.17.26",
       "resolved": "https://registry.npmjs.org/@types/express-serve-static-core/-/express-serve-static-core-4.17.26.tgz",
@@ -9331,6 +9386,11 @@
       "resolved": "https://registry.npmjs.org/@types/long/-/long-4.0.1.tgz",
       "integrity": "sha512-5tXH6Bx/kNGd3MgffdmP4dy2Z+G4eaXw0SE81Tq3BNadtnMR5/ySMzX4SLEzHJzSmPNn4HIdpQsBvXMUykr58w=="
     },
+    "@types/mime": {
+      "version": "1.3.2",
+      "resolved": "https://registry.npmjs.org/@types/mime/-/mime-1.3.2.tgz",
+      "integrity": "sha512-YATxVxgRqNH6nHEIsvg6k2Boc1JHI9ZbH5iWFFv/MTkchz3b1ieGDa5T0a9RznNdI0KhVbdbWSN+KWWrQZRxTw=="
+    },
     "@types/node": {
       "version": "16.11.11",
       "resolved": "https://registry.npmjs.org/@types/node/-/node-16.11.11.tgz",
@@ -9352,6 +9412,15 @@
       "resolved": "https://registry.npmjs.org/@types/range-parser/-/range-parser-1.2.4.tgz",
       "integrity": "sha512-EEhsLsD6UsDM1yFhAvy0Cjr6VwmpMWqFBCb9w07wVugF7w9nfajxLuVmngTIpgS6svCnm6Vaw+MZhoDCKnOfsw=="
     },
+    "@types/serve-static": {
+      "version": "1.13.10",
+      "resolved": "https://registry.npmjs.org/@types/serve-static/-/serve-static-1.13.10.tgz",
+      "integrity": "sha512-nCkHGI4w7ZgAdNkrEu0bv+4xNV/XDqW+DydknebMOQwkpDGx8G+HTlj7R7ABI8i8nKxVw0wtKPi1D+lPOkh4YQ==",
+      "requires": {
+        "@types/mime": "^1",
+        "@types/node": "*"
+      }
+    },
     "@types/stack-utils": {
       "version": "2.0.1",
       "resolved": "https://registry.npmjs.org/@types/stack-utils/-/stack-utils-2.0.1.tgz",

third_party/pyth/p2w-terra-relay/src/helpers.ts

@@ -1,3 +1,5 @@
+import assert = require("assert");
+
 ////////////////////////////////// Start of Logger Stuff //////////////////////////////////////
 
 export let logger: any;
@@ -58,7 +60,8 @@ export function initLogger() {
 ////////////////////////////////// Start of PYTH Stuff //////////////////////////////////////
 
 /*
-  // Pyth PriceAttestation messages are defined in wormhole/ethereum/contracts/pyth/PythStructs.sol
+  // Pyth messages are defined in whitepapers/0007_pyth_over_wormhole.md
+
   // The Pyth smart contract stuff is in terra/contracts/pyth-bridge
 
   struct Ema {
@@ -108,9 +111,43 @@ export function initLogger() {
 141 u8        corpAct
 142 u64       timestamp
 
+In version 2 prices are sent in batch with the following structure:
+
+  struct BatchPriceAttestation {
+      uint32 magic; // constant "P2WH"
+      uint16 version;
+
+      // PayloadID uint8 = 2
+      uint8 payloadId;
+
+      // number of attestations 
+      uint16 nAttestations;
+
+      // Length of each price attestation in bytes
+      //
+      // This field is provided for forwards compatibility. Fields in future may be added in
+      // an append-only way allowing for parsers to continue to work by parsing only up to
+      // the fields they know, leaving unread input in the buffer. Code may still need to work
+      // with the full size of the value however, such as when iterating over lists of attestations,
+      // for these use-cases the structure size is included as a field.
+      //
+      // attestation_size >= 150
+      uint16 attestationSize;
+      
+      priceAttestations: PriceAttestation[]
+  }
+
+0   uint32    magic // constant "P2WH"
+4   u16       version
+6   u8        payloadId // 2
+7   u16       n_attestations
+9   u16       attestation_size // >= 150
+11  ..        price_attestation (Size: attestation_size x [n_attestations])
+
 */
 
-export const PYTH_PRICE_ATTESTATION_LENGTH: number = 150;
+export const PYTH_PRICE_ATTESTATION_MIN_LENGTH: number = 150;
+export const PYTH_BATCH_PRICE_ATTESTATION_MIN_LENGTH: number = 11 + PYTH_PRICE_ATTESTATION_MIN_LENGTH;
 
 export type PythEma = {
   value: BigInt;
@@ -135,8 +172,32 @@ export type PythPriceAttestation = {
   timestamp: BigInt;
 };
 
+export type PythBatchPriceAttestation = {
+  magic: number;
+  version: number; 
+  payloadId: number;
+  nAttestations: number;
+  attestationSize: number;
+  priceAttestations: PythPriceAttestation[];
+}
+
 export const PYTH_MAGIC: number = 0x50325748;
 
+function isPyth(payload: Buffer): boolean {
+  if (payload.length < 4) return false;
+  if (
+    payload[0] === 80 &&
+    payload[1] === 50 &&
+    payload[2] === 87 &&
+    payload[3] === 72
+  ) {
+    // The numbers correspond to "P2WH"
+    return true;
+  }
+
+  return false;
+}
+
 export function parsePythPriceAttestation(arr: Buffer): PythPriceAttestation {
   return {
     magic: arr.readUInt32BE(0),
@@ -164,6 +225,73 @@ export function parsePythPriceAttestation(arr: Buffer): PythPriceAttestation {
   };
 }
 
+export function parsePythBatchPriceAttestation(arr: Buffer): PythBatchPriceAttestation {
+  if (!isPyth(arr)) {
+    throw new Error("Cannot parse payload. Header mismatch: This is not a Pyth 2 Wormhole message");
+  }
+
+  if (arr.length < PYTH_BATCH_PRICE_ATTESTATION_MIN_LENGTH) {
+    throw new Error(
+        "Cannot parse payload. Payload length is wrong: length: " +
+        arr.length +
+        ", expected length to be at least:" +
+        PYTH_BATCH_PRICE_ATTESTATION_MIN_LENGTH
+    );
+  }
+
+  const magic = arr.readUInt32BE(0);
+  const version = arr.readUInt16BE(4);
+  const payloadId = arr[6];
+  const nAttestations = arr.readUInt16BE(7);
+  const attestationSize = arr.readUInt16BE(9);
+
+  if (attestationSize < PYTH_PRICE_ATTESTATION_MIN_LENGTH) {
+    throw new Error(
+      `Cannot parse payload. Size of attestation ${attestationSize} is less than V2 length ${PYTH_PRICE_ATTESTATION_MIN_LENGTH}`
+    );
+  }
+
+  let priceAttestations: PythPriceAttestation[] = []
+
+  let offset = 11;
+  for (let i = 0; i < nAttestations; i += 1) {
+    priceAttestations.push(parsePythPriceAttestation(arr.subarray(offset, offset + attestationSize)));
+    offset += attestationSize;
+  }
+
+  return {
+      magic,
+      version,
+      payloadId,
+      nAttestations,
+      attestationSize,
+      priceAttestations
+    }
+}
+
+// Returns a hash of all priceIds within the batch, it can be used to identify whether there is a
+// new batch with exact same symbols (and ignore the old one)
+export function getBatchAttestationHashKey(batchAttestation: PythBatchPriceAttestation): string {
+  const priceIds: string[] = batchAttestation.priceAttestations.map((priceAttestation) => priceAttestation.priceId);
+  priceIds.sort();
+
+  return priceIds.join('#');
+}
+
+export function getBatchSummary(batchAttestation: PythBatchPriceAttestation): string {
+  let abstractRepresentation = {
+    "num_attestations": batchAttestation.nAttestations,
+    "prices": batchAttestation.priceAttestations.map((priceAttestation) => {
+        return {
+          "price_id": priceAttestation.priceId,
+          "price": computePrice(priceAttestation.price, priceAttestation.exponent),
+          "conf": computePrice(priceAttestation.confidenceInterval, priceAttestation.exponent)
+        }
+    })
+  }
+  return JSON.stringify(abstractRepresentation)
+}
+
 ////////////////////////////////// Start of Other Helpful Stuff //////////////////////////////////////
 
 export function sleep(ms: number) {

third_party/pyth/p2w-terra-relay/src/listen.ts

@@ -147,101 +147,51 @@ async function processVaa(vaaBytes: string) {
   // );
 
   // logger.debug("listen:processVaa: parsedVAA: %o", parsedVAA);
+  
+  let batchAttestation;
+
+  try {
+    batchAttestation = helpers.parsePythBatchPriceAttestation(Buffer.from(parsedVAA.payload));
+  } catch (e: any) {
+    logger.error(e, e.stack);
+    logger.error("Parsing failed. Dropping vaa: %o", parsedVAA)
+    return;
+  }
 
-  if (isPyth(parsedVAA.payload)) {
-    if (parsedVAA.payload.length < helpers.PYTH_PRICE_ATTESTATION_LENGTH) {
-      logger.error(
-        "dropping vaa because the payload length is wrong: length: " +
-          parsedVAA.payload.length +
-          ", expected length:",
-        helpers.PYTH_PRICE_ATTESTATION_LENGTH + ", vaa: %o",
-        parsedVAA
-      );
-      return;
-    }
+  let isAnyPriceNew = batchAttestation.priceAttestations.some(priceAttestation => {
+    const key = priceAttestation.priceId;
+    let lastSeqNum = seqMap.get(key);
+    return lastSeqNum === undefined || lastSeqNum < parsedVAA.sequence;
+  })
 
-    let pa = helpers.parsePythPriceAttestation(Buffer.from(parsedVAA.payload));
-    // logger.debug("listen:processVaa: price attestation: %o", pa);
+  if (!isAnyPriceNew) {
+    logger.debug("For all prices there exists an update with newer sequence number. batch price attestation: %o", batchAttestation);
+    return;
+  }
 
-    let key = pa.priceId;
-    let lastSeqNum = seqMap.get(key);
-    if (lastSeqNum) {
-      if (lastSeqNum >= parsedVAA.sequence) {
-        logger.debug(
-          "ignoring duplicate: emitter: [" +
-            parsedVAA.emitter_chain +
-            ":" +
-            uint8ArrayToHex(parsedVAA.emitter_address) +
-            "], productId: [" +
-            pa.productId +
-            "], priceId: [" +
-            pa.priceId +
-            "], seqNum: " +
-            parsedVAA.sequence
-        );
-        return;
-      }
-    }
+  for (let priceAttestation of batchAttestation.priceAttestations) {
+    const key = priceAttestation.priceId;
 
-    seqMap.set(key, parsedVAA.sequence);
-
-    logger.info(
-      "received: emitter: [" +
-        parsedVAA.emitter_chain +
-        ":" +
-        uint8ArrayToHex(parsedVAA.emitter_address) +
-        "], seqNum: " +
-        parsedVAA.sequence +
-        ", productId: [" +
-        pa.productId +
-        "], priceId: [" +
-        pa.priceId +
-        "], priceType: " +
-        pa.priceType +
-        ", price: " +
-        pa.price +
-        ", exponent: " +
-        pa.exponent +
-        ", confidenceInterval: " +
-        pa.confidenceInterval +
-        ", timeStamp: " +
-        pa.timestamp +
-        ", computedPrice: " +
-        helpers.computePrice(pa.price, pa.exponent) +
-        " +/-" +
-        helpers.computePrice(pa.confidenceInterval, pa.exponent)
-      // +
-      // ", payload: [" +
-      // uint8ArrayToHex(parsedVAA.payload) +
-      // "]"
-    );
-
-    metrics.incIncoming();
-    if (!listenOnly) {
-      logger.debug("posting to worker");
-      await postEvent(vaaBytes, pa, parsedVAA.sequence, receiveTime);
+    let lastSeqNum = seqMap.get(key);
+    if (lastSeqNum === undefined || lastSeqNum < parsedVAA.sequence) {
+      seqMap.set(key, parsedVAA.sequence);
     }
-  } else {
-    logger.debug(
-      "dropping non-pyth vaa, payload type " +
-        parsedVAA.payload[0] +
-        ", vaa: %o",
-      parsedVAA
-    );
   }
-}
 
-function isPyth(payload: Buffer): boolean {
-  if (payload.length < 4) return false;
-  if (
-    payload[0] === 80 &&
-    payload[1] === 50 &&
-    payload[2] === 87 &&
-    payload[3] === 72
-  ) {
-    // P2WH
-    return true;
-  }
+  logger.info(
+    "received: emitter: [" +
+      parsedVAA.emitter_chain +
+      ":" +
+      uint8ArrayToHex(parsedVAA.emitter_address) +
+      "], seqNum: " +
+      parsedVAA.sequence +
+      ", Batch Summary: " +
+      helpers.getBatchSummary(batchAttestation)
+  );
 
-  return false;
+  metrics.incIncoming();
+  if (!listenOnly) {
+    logger.debug("posting to worker");
+    await postEvent(vaaBytes, batchAttestation, parsedVAA.sequence, receiveTime);
+  }
 }

third_party/pyth/p2w-terra-relay/src/relay/main.ts

@@ -44,13 +44,12 @@ export async function relay(
 }
 
 export async function query(
-  productIdStr: string,
   priceIdStr: string
 ): Promise<any> {
   let result: any;
   try {
     let terraData = connectToTerra();
-    result = await queryTerra(terraData, productIdStr, priceIdStr);
+    result = await queryTerra(terraData, priceIdStr);
   } catch (e) {
     logger.error("query failed: %o", e);
     result = "Error: unhandled exception";

third_party/pyth/p2w-terra-relay/src/relay/terra.ts

@@ -152,18 +152,12 @@ export async function relayTerra(
 
 export async function queryTerra(
   connectionData: TerraConnectionData,
-  productIdStr: string,
   priceIdStr: string
 ) {
-  const encodedProductId = fromUint8Array(hexToUint8Array(productIdStr));
   const encodedPriceId = fromUint8Array(hexToUint8Array(priceIdStr));
 
   logger.info(
-    "Querying terra for price info for productId [" +
-      productIdStr +
-      "], encoded as [" +
-      encodedProductId +
-      "], priceId [" +
+    "Querying terra for price info for priceId [" +
       priceIdStr +
       "], encoded as [" +
       encodedPriceId +
@@ -182,7 +176,6 @@ export async function queryTerra(
     connectionData.contractAddress,
     {
       price_info: {
-        product_id: encodedProductId,
         price_id: encodedPriceId,
       },
     }

third_party/pyth/p2w-terra-relay/src/rest.ts

@@ -31,10 +31,9 @@ export async function run() {
     });
 
     app.get(
-      "/queryterra/:product_id/:price_id",
+      "/queryterra/:price_id",
       async (req: Request, res: Response) => {
         let result = await getPriceData(
-          req.params.product_id,
           req.params.price_id
         );
         res.json(result);
@@ -42,7 +41,7 @@ export async function run() {
     );
 
     app.get("/", (req: Request, res: Response) =>
-      res.json(["/status", "/queryterra/<product_id>/<price_id>"])
+      res.json(["/status", "/queryterra/<price_id>"])
     );
   })();
 }

third_party/pyth/p2w-terra-relay/src/worker.ts

@@ -15,18 +15,18 @@ let conditionTimeout = 20000;
 
 type PendingPayload = {
   vaa_bytes: string;
-  pa: helpers.PythPriceAttestation;
+  batchAttestation: helpers.PythBatchPriceAttestation;
   receiveTime: Date;
   seqNum: number;
 };
 
-let pendingMap = new Map<string, PendingPayload>(); // The key to this is price_id. Note that Map maintains insertion order, not key order.
+let pendingMap = new Map<string, PendingPayload>(); // The key to this is hash of price_ids in the batch attestation. Note that Map maintains insertion order, not key order.
 
 type ProductData = {
   key: string;
   lastTimePublished: Date;
   numTimesPublished: number;
-  lastPa: helpers.PythPriceAttestation;
+  lastBatchAttestation: helpers.PythBatchPriceAttestation;
   lastResult: any;
 };
 
@@ -35,7 +35,7 @@ type CurrentEntry = {
   currObj: ProductData;
 };
 
-let productMap = new Map<string, ProductData>(); // The key to this is price_id
+let productMap = new Map<string, ProductData>(); // The key to this is hash of price_ids in the batch attestation.
 
 let connectionData: main.ConnectionData;
 let metrics: PromHelper;
@@ -233,11 +233,11 @@ async function getPendingEventsAlreadyLocked(
   while (pendingMap.size !== 0 && currObjs.length < maxPerBatch) {
     const first = pendingMap.entries().next();
     logger.debug("processing event with key [" + first.value[0] + "]");
-    const pendingValue = first.value[1];
-    let pendingKey = pendingValue.pa.priceId;
+    const pendingValue: PendingPayload = first.value[1];
+    let pendingKey = helpers.getBatchAttestationHashKey(pendingValue.batchAttestation);
     let currObj = productMap.get(pendingKey);
     if (currObj) {
-      currObj.lastPa = pendingValue.pa;
+      currObj.lastBatchAttestation = pendingValue.batchAttestation;
       currObj.lastTimePublished = new Date();
       productMap.set(pendingKey, currObj);
       logger.debug(
@@ -257,7 +257,7 @@ async function getPendingEventsAlreadyLocked(
       );
       currObj = {
         key: pendingKey,
-        lastPa: pendingValue.pa,
+        lastBatchAttestation: pendingValue.batchAttestation,
         lastTimePublished: new Date(),
         numTimesPublished: 0,
         lastResult: "",
@@ -458,17 +458,11 @@ async function finalizeEventsAlreadyLocked(
     );
 
     logger.info(
-      "complete: priceId: " +
-        currEntry.pa.priceId +
-        ", seqNum: " +
+      "complete:" +
+        "seqNum: " +
         currEntry.seqNum +
-        ", price: " +
-        helpers.computePrice(currEntry.pa.price, currEntry.pa.exponent) +
-        ", ci: " +
-        helpers.computePrice(
-          currEntry.pa.confidenceInterval,
-          currEntry.pa.exponent
-        ) +
+        ", price_ids: " +
+        helpers.getBatchSummary(currEntry.batchAttestation) +
         ", rcv2SendBegin: " +
         (sendTime.getTime() - currEntry.receiveTime.getTime()) +
         ", rcv2SendComplete: " +
@@ -503,18 +497,17 @@ async function finalizeEventsAlreadyLocked(
 
 export async function postEvent(
   vaaBytes: any,
-  pa: helpers.PythPriceAttestation,
+  batchAttestation: helpers.PythBatchPriceAttestation,
   sequence: number,
   receiveTime: Date
 ) {
   let event: PendingPayload = {
     vaa_bytes: uint8ArrayToHex(vaaBytes),
-    pa: pa,
+    batchAttestation: batchAttestation,
     receiveTime: receiveTime,
     seqNum: sequence,
   };
-  let pendingKey = pa.priceId;
-  // pendingKey = pendingKey + ":" + sequence;
+  let pendingKey = helpers.getBatchAttestationHashKey(batchAttestation);
   await mutex.runExclusive(() => {
     logger.debug("posting event with key [" + pendingKey + "]");
     pendingMap.set(pendingKey, event);
@@ -537,13 +530,7 @@ export async function getStatus() {
       }
 
       let item: object = {
-        product_id: value.lastPa.productId,
-        price_id: value.lastPa.priceId,
-        price: helpers.computePrice(value.lastPa.price, value.lastPa.exponent),
-        ci: helpers.computePrice(
-          value.lastPa.confidenceInterval,
-          value.lastPa.exponent
-        ),
+        summary: helpers.getBatchSummary(value.lastBatchAttestation),
         num_times_published: value.numTimesPublished,
         last_time_published: value.lastTimePublished.toISOString(),
         result: value.lastResult,
@@ -559,12 +546,11 @@ export async function getStatus() {
 
 // Note that querying the contract does not update the sequence number, so we don't need to be locked.
 export async function getPriceData(
-  productId: string,
   priceId: string
 ): Promise<any> {
   let result: any;
   // await mutex.runExclusive(async () => {
-  result = await main.query(productId, priceId);
+  result = await main.query(priceId);
   // });
 
   return result;