Merge remote-tracking branch 'origin/master' into lash/session-nonce-queue

Atomic nonce queue db sessions

See merge request grassrootseconomics/cic-internal-integration!48

apps/cic-cache/cic_cache/runnable/tracker.py

@@ -312,7 +312,7 @@ class Tracker:
         session.close()
 
 (provider, w3) = web3_constructor()
-trust = config.get('CIC_TRUST_ADDRESS', []).split(",")
+trust = config.get('CIC_TRUST_ADDRESS', "").split(",")
 chain_spec = args.i
 
 try:

apps/cic-eth/cic_eth/api/api_task.py

@@ -490,7 +490,7 @@ class Api:
             queue=self.queue,
             )
         if self.callback_param != None:
-            s_brief.link(self.callback_success).on_error(self.callback_error)
+            s_assemble.link(self.callback_success).on_error(self.callback_error)
 
         t = None
         if external_task != None:
@@ -515,8 +515,7 @@ class Api:
             c = celery.chain(s_external_get, s_external_process)
             t = celery.chord([s_local, c])(s_brief)
         else:
-            s_local.link(s_brief)
-            t = s_local.apply_async()
+            t = s_local.apply_sync()
 
         return t
 

apps/cic-eth/cic_eth/eth/account.py

@@ -204,8 +204,9 @@ def register(self, account_address, chain_str, writer_address=None):
     txf = AccountTxFactory(writer_address, c)
 
     tx_add = txf.add(account_address, chain_spec, session=session)
+    
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_account_data', session=session)
     session.close()
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_account_data')
 
     gas_budget = tx_add['gas'] * tx_add['gasPrice']
 
@@ -241,12 +242,14 @@ def gift(self, account_address, chain_str):
     c = RpcClient(chain_spec, holder_address=account_address)
     txf = AccountTxFactory(account_address, c)
 
-    tx_add = txf.gift(account_address, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_gift_data')
+    session = SessionBase.create_session()
+    tx_add = txf.gift(account_address, chain_spec, session=session)
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_add, chain_str, queue, 'cic_eth.eth.account.cache_gift_data', session=session)
+    session.close()
 
     gas_budget = tx_add['gas'] * tx_add['gasPrice']
 
-    logg.debug('register user tx {}'.format(tx_hash_hex))
+    logg.debug('gift user tx {}'.format(tx_hash_hex))
     s = create_check_gas_and_send_task(
             [tx_signed_raw_hex],
             chain_str,

apps/cic-eth/cic_eth/eth/token.py

@@ -46,6 +46,7 @@ class TokenTxFactory(TxFactory):
             spender_address,
             amount,
             chain_spec,
+            session=None,
             ):
         """Create an ERC20 "approve" transaction
 
@@ -73,7 +74,7 @@ class TokenTxFactory(TxFactory):
             'gas': source_token_gas,
             'gasPrice': self.gas_price,
             'chainId': chain_spec.chain_id(),
-            'nonce': self.next_nonce(),
+            'nonce': self.next_nonce(session=session),
             })
         return tx_approve
 
@@ -84,6 +85,7 @@ class TokenTxFactory(TxFactory):
         receiver_address,
         value,
         chain_spec,
+        session=None,
         ):
         """Create an ERC20 "transfer" transaction
 
@@ -112,7 +114,7 @@ class TokenTxFactory(TxFactory):
                     'gas': source_token_gas,
                     'gasPrice': self.gas_price,
                     'chainId': chain_spec.chain_id(),
-                    'nonce': self.next_nonce(),
+                    'nonce': self.next_nonce(session=session),
                 })
         return tx_transfer
 
@@ -244,9 +246,11 @@ def transfer(self, tokens, holder_address, receiver_address, value, chain_str):
     c = RpcClient(chain_spec, holder_address=holder_address)
 
     txf = TokenTxFactory(holder_address, c)
-
-    tx_transfer = txf.transfer(t['address'], receiver_address, value, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_transfer')
+    
+    session = SessionBase.create_session()
+    tx_transfer = txf.transfer(t['address'], receiver_address, value, chain_spec, session=session)
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_transfer', session=session)
+    session.close()
     
     gas_budget = tx_transfer['gas'] * tx_transfer['gasPrice']
 
@@ -299,8 +303,10 @@ def approve(self, tokens, holder_address, spender_address, value, chain_str):
 
     txf = TokenTxFactory(holder_address, c)
 
-    tx_transfer = txf.approve(t['address'], spender_address, value, chain_spec)
-    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_approve')
+    session = SessionBase.create_session()
+    tx_transfer = txf.approve(t['address'], spender_address, value, chain_spec, session=session)
+    (tx_hash_hex, tx_signed_raw_hex) = sign_and_register_tx(tx_transfer, chain_str, queue, cache_task='cic_eth.eth.token.otx_cache_approve', session=session)
+    session.close()
     
     gas_budget = tx_transfer['gas'] * tx_transfer['gasPrice']
 
@@ -459,6 +465,7 @@ def cache_approve_data(
     return (tx_hash_hex, cache_id)
 
 
+# TODO: Move to dedicated metadata package
 class ExtendedTx:
 
     _default_decimals = 6

apps/cic-eth/cic_eth/eth/tx.py

@@ -387,6 +387,7 @@ def send(self, txs, chain_str):
 
 
 # TODO: if this method fails the nonce will be out of sequence. session needs to be extended to include the queue create, so that nonce is rolled back if the second sql query fails. Better yet, split each state change into separate tasks.
+# TODO: method is too long, factor out code for clarity
 @celery_app.task(bind=True, throws=(web3.exceptions.TransactionNotFound,), base=CriticalWeb3Task)
 def refill_gas(self, recipient_address, chain_str):
     """Executes a native token transaction to fund the recipient's gas expenditures.
@@ -409,8 +410,8 @@ def refill_gas(self, recipient_address, chain_str):
     q = q.filter(TxCache.from_value!=0)
     q = q.filter(TxCache.recipient==recipient_address)
     c = q.count()
-    session.close()
     if c > 0:
+        session.close()
         raise AlreadyFillingGasError(recipient_address)
 
     queue = self.request.delivery_info['routing_key']
@@ -420,7 +421,7 @@ def refill_gas(self, recipient_address, chain_str):
     logg.debug('refill gas from provider address {}'.format(c.gas_provider()))
     default_nonce = c.w3.eth.getTransactionCount(c.gas_provider(), 'pending')
     nonce_generator = NonceOracle(c.gas_provider(), default_nonce)
-    nonce = nonce_generator.next()
+    nonce = nonce_generator.next(session=session)
     gas_price = c.gas_price()
     gas_limit = c.default_gas_limit
     refill_amount = c.refill_amount()
@@ -449,7 +450,9 @@ def refill_gas(self, recipient_address, chain_str):
         tx_hash_hex,
         tx_send_gas_signed['raw'],
         chain_str,
+        session=session,
         )
+    session.close()
 
     s_tx_cache = celery.signature(
             'cic_eth.eth.tx.cache_gas_refill_data',
@@ -492,8 +495,8 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
     q = session.query(Otx)
     q = q.filter(Otx.tx_hash==txold_hash_hex)
     otx = q.first()
-    session.close()
     if otx == None:
+        session.close()
         raise NotLocalTxError(txold_hash_hex)
 
     chain_spec = ChainSpec.from_chain_str(chain_str)
@@ -528,8 +531,10 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
         tx_hash_hex,
         tx_signed_raw_hex,
         chain_str,
+        session=session,
             )
-    TxCache.clone(txold_hash_hex, tx_hash_hex)
+    TxCache.clone(txold_hash_hex, tx_hash_hex, session=session)
+    session.close()
 
     s = create_check_gas_and_send_task(
             [tx_signed_raw_hex],
@@ -546,6 +551,13 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_str, gas=None, default_fa
 
 @celery_app.task(bind=True, throws=(web3.exceptions.TransactionNotFound,), base=CriticalWeb3Task)
 def sync_tx(self, tx_hash_hex, chain_str):
+    """Force update of network status of a simgle transaction
+
+    :param tx_hash_hex: Transaction hash
+    :type tx_hash_hex: str, 0x-hex
+    :param chain_str: Chain spec string representation
+    :type chain_str: str
+    """
 
     queue = self.request.delivery_info['routing_key']
 

apps/cic-eth/cic_eth/ext/tx.py

@@ -149,9 +149,6 @@ def tx_collate(tx_batches, chain_str, offset, limit, newest_first=True):
     txs_by_block = {}
     chain_spec = ChainSpec.from_chain_str(chain_str)
 
-    if isinstance(tx_batches, dict):
-        tx_batches = [tx_batches]
-
     for b in tx_batches:
         for v in b.values():
             tx = None

apps/cic-eth/cic_eth/runnable/daemons/tasker.py

@@ -40,8 +40,6 @@ from cic_eth.callbacks import redis
 from cic_eth.db.models.base import SessionBase
 from cic_eth.db.models.otx import Otx
 from cic_eth.db import dsn_from_config
-from cic_eth.ext import tx
-from cic_eth.ext import address
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()

apps/cic-meta/.gitignore

@@ -1,5 +1,6 @@
 node_modules
 dist
 dist-web
+dist-server
 scratch
 tests

apps/cic-meta/CHANGELOG

@@ -1,6 +1,7 @@
 * 0.0.7-pending
 	- Add immutable content
 	- Add db lock on server
+	- Add ArgPair and KeyStore to src exports
 * 0.0.6
 	- Add server build
 * 0.0.5

apps/cic-meta/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "cic-client-meta",
-	"version": "0.0.7-alpha.2",
+	"version": "0.0.7-alpha.3",
 	"description": "Signed CRDT metadata graphs for the CIC network",
 	"main": "dist/index.js",
 	"types": "dist/index.d.ts",
@@ -32,6 +32,12 @@
 		"webpack-cli": "^4.2.0"
 	},
 	"author": "Louis Holbrook <dev@holbrook.no>",
+	"contributors": [
+		{
+			"name": "Spencer Ofwiti",
+			"email": "maxspencer56@gmail.com"
+		}
+	],
 	"license": "GPL-3.0-or-later",
 	"engines": {
 		"node": "~15.3.0"

apps/cic-meta/scripts/server/server.ts

@@ -101,6 +101,18 @@ function parseDigest(url) {
 
 async function processRequest(req, res) {
 	let digest = undefined;
+	const headers = {
+		"Access-Control-Allow-Origin": "*",
+		"Access-Control-Allow-Methods": "OPTIONS, POST, GET, PUT",
+		"Access-Control-Max-Age": 2592000, // 30 days
+		"Access-Control-Allow-Headers": 'Access-Control-Allow-Origin, Content-Type, x-cic-automerge'
+	};
+
+	if (req.method === "OPTIONS") {
+		res.writeHead(200, headers);
+		res.end();
+		return;
+	}
 
 	if (!['PUT', 'GET', 'POST'].includes(req.method)) {
 		res.writeHead(405, {"Content-Type": "text/plain"});
@@ -201,6 +213,7 @@ async function processRequest(req, res) {
 
 		const responseContentLength = (new TextEncoder().encode(content)).length;
 		res.writeHead(200, {
+			"Access-Control-Allow-Origin": "*",
 			"Content-Type": contentType,
 			"Content-Length": responseContentLength,
 		});

apps/cic-meta/src/index.ts

@@ -1,5 +1,5 @@
-export { PGPSigner, PGPKeyStore } from './auth';
-export { Envelope, Syncable } from './sync';
+export { PGPSigner, PGPKeyStore, Signer, KeyStore } from './auth';
+export { ArgPair,  Envelope, Syncable } from './sync';
 export { User } from './assets/user';
 export { Phone } from './assets/phone';
 export { Config } from './config';

apps/cic-meta/src/sync.ts

@@ -93,7 +93,7 @@ function orderDict(src) {
 	return dst;
 }
 
-class Syncable implements JSONSerializable, Authoritative {
+class Syncable implements JSONSerializable, Authoritative, Signable {
 
 	id:		string
 	timestamp:	number

ci_templates/.cic-template.yml

@@ -40,11 +40,12 @@ before_script:
       - echo "$CI_REGISTRY_PASSWORD" | docker login -u "$CI_REGISTRY_USER" $CI_REGISTRY --password-stdin
     variables:
         CI_DEBUG_TRACE: "true"
-        IMAGE_TAG: $CI_REGISTRY_IMAGE/$APP_NAME:$CI_COMMIT_BRANCH-$CI_COMMIT_SHORT_SHA
+        IMAGE_TAG_BASE: $CI_REGISTRY_IMAGE/$APP_NAME:$CI_COMMIT_BRANCH-$CI_COMMIT_SHORT_SHA
         LATEST_TAG: $CI_REGISTRY_IMAGE/$APP_NAME:latest
     script:
+      - export IMAGE_TAG="$IMAGE_TAG_BASE-$(date +%F.%H%M%S)"
       - docker build -t $IMAGE_TAG -f $DOCKERFILE_PATH .
-      - docker push $IMAGE_TAG
+      - docker push "$IMAGE_TAG
       - docker tag $IMAGE_TAG $LATEST_TAG
       - docker push $LATEST_TAG
     rules: