Explicit IPFS peering between discprov and creator nodes

creator-node/src/routes/files.js

@@ -28,6 +28,12 @@ module.exports = function (app) {
  return successResponse({ 'metadataMultihash': multihash })
  }))
 
+ app.get('/ipfs_peer_info', handleResponse(async (req, res) => {
+ const ipfs = req.app.get('ipfsAPI')
+ let ipfsIDObj = await ipfs.id()
+ return successResponse(ipfsIDObj)
+ }))
+
  // Serve audius network IPFS segments
  // This route does not handle responses by design, so we can pipe the gateway response
  app.get('/ipfs/:multihash', async (req, res) => {

discovery-provider/.env

@@ -5,6 +5,7 @@ audius_redis_url=redis://redis-server:6379/00
 audius_db_url=postgresql+psycopg2://postgres:postgres@db:5432/audius_discovery
 audius_ipfs_host=docker.for.mac.localhost
 audius_ipfs_port=6001
-audius_ipfs_gateway_hosts=http://docker.for.mac.localhost:8080
+audius_ipfs_gateway_hosts=
 audius_discprov_block_processing_window=100
+audius_discprov_peer_refresh_interval=10
 WAIT_HOSTS=db:5432,redis-server:6379,docker.for.mac.localhost:6001

discovery-provider/default_config.ini

@@ -4,6 +4,7 @@ loglevel_flask = DEBUG
 loglevel_celery = INFO
 block_processing_window = 50
 blacklist_block_processing_window = 600
+peer_refresh_interval = 3000
 
 [flask]
 debug = true
@@ -43,4 +44,4 @@ registry = 0x2999e02829DC711B9254187962ba44d1fFcf5481
 
 [delegate]
 owner_wallet =
-private_key =
+private_key =

discovery-provider/src/tasks/index.py

@@ -3,7 +3,7 @@
 from src.models import Block, User, Track, Repost, Follow, Playlist, Save
 from src.tasks.celery_app import celery
 from src.tasks.tracks import track_state_update
-from src.tasks.users import user_state_update # pylint: disable=E0611,E0001
+from src.tasks.users import user_state_update, get_ipfs_info_from_cnode_endpoint # pylint: disable=E0611,E0001
 from src.tasks.social_features import social_feature_state_update
 from src.tasks.playlists import playlist_state_update
 from src.tasks.user_library import user_library_state_update
@@ -391,10 +391,53 @@ def revert_blocks(self, db, revert_blocks_list):
  if rebuild_user_index:
  session.execute("REFRESH MATERIALIZED VIEW user_lexeme_dict")
 
+######## IPFS PEER REFRESH ########
+def refresh_peer_connections(task_context):
+ db = task_context.db
+ ipfs_client = task_context.ipfs_client
+ redis = task_context.redis
+ interval = int(task_context.shared_config["discprov"]["peer_refresh_interval"])
+ with db.scoped_session() as session:
+ db_cnode_endpts = (
+ session.query(
+ User.creator_node_endpoint).filter(
+ User.creator_node_endpoint != None, User.is_current == True
+ ).distinct()
+ )
+
+ cnode_endpoints = {}
+ # Generate dictionary of unique creator node endpoints
+ for entry in db_cnode_endpts:
+ for cnode_user_set in entry:
+ cnode_entries = cnode_user_set.split(',')
+ for cnode_url in cnode_entries:
+ if cnode_url == '':
+ continue
+ cnode_endpoints[cnode_url] = True
+
+ # Query ipfs information for each cnode endpoint
+ multiaddr_info = {}
+ for cnode_url in cnode_endpoints:
+ stored_in_redis = redis.get(cnode_url)
+ if stored_in_redis is not None:
+ continue
+ if cnode_url == '':
+ continue
+
+ try:
+ multiaddr_info[cnode_url] = get_ipfs_info_from_cnode_endpoint(cnode_url)
+ except Exception as e: # pylint: disable=broad-except
+ # Handle error in retrieval by not peering this node
+ logger.warning('Error retrieving info for %s, %s', cnode_url, str(e))
+ multiaddr_info[cnode_url] = None
+
+ for key in multiaddr_info:
+ if multiaddr_info[key] is not None:
+ # Peer nodes
+ ipfs_client.connect_peer(multiaddr_info[key])
+ redis.set(key, multiaddr_info[key], ex=interval)
 
 ######## CELERY TASKS ########
-
-
 @celery.task(name="update_discovery_provider", bind=True)
 def update_task(self):
  # Cache custom task class properties
@@ -411,6 +454,9 @@ def update_task(self):
  # Attempt to acquire lock - do not block if unable to acquire
  have_lock = update_lock.acquire(blocking=False)
  if have_lock:
+ # Refresh all IPFS peer connections
+ refresh_peer_connections(self)
+
  logger.debug(f"index.py | update_task | Acquired disc_prov_lock")
  initialize_blocks_table_if_necessary(db)
 

discovery-provider/src/tasks/index_blacklist.py

@@ -133,7 +133,7 @@ def index_blocks(self, db, blocks_list):
  )
 
  if num_blocks > 0:
- logger.warning(f"IPLDBLACKLIST | Indexed {num_blocks} blocks")
+ logger.info(f"IPLDBLACKLIST | Indexed {num_blocks} blocks")
 
 
 def revert_blocks(self, db, revert_blocks_list):

discovery-provider/src/tasks/users.py

@@ -1,6 +1,8 @@
 import logging
+from urllib.parse import urljoin
 from datetime import datetime
 from sqlalchemy.orm.session import make_transient
+import requests
 from src import contract_addresses
 from src.utils import helpers
 from src.models import User, BlacklistedIPLD
@@ -211,8 +213,34 @@ def get_metadata_overrides_from_ipfs(session, update_task, user_record):
  if ipld_blacklist_entry:
  return None
 
+ # Manually peer with user creator nodes
+ update_ipfs_peers_from_user_endpoint(
+ update_task,
+ user_record.creator_node_endpoint)
+
  user_metadata = update_task.ipfs_client.get_metadata(
  user_record.metadata_multihash,
  user_metadata_format)
 
  return user_metadata
+
+def get_ipfs_info_from_cnode_endpoint(url):
+ id_url = urljoin(url, 'ipfs_peer_info')
+ resp = requests.get(id_url, timeout=5)
+ json_resp = resp.json()
+ ipfs_multiaddr_entries = json_resp['addresses']
+ for multiaddr in ipfs_multiaddr_entries:
+ if ('127.0.0.1' not in multiaddr) and ('ip6' not in multiaddr):
+ return multiaddr
+ raise Exception('Failed to find valid multiaddr')
+
+def update_ipfs_peers_from_user_endpoint(update_task, cnode_url_list):
+ if cnode_url_list is None:
+ return
+ redis = update_task.redis
+ cnode_entries = cnode_url_list.split(',')
+ interval = int(update_task.shared_config["discprov"]["peer_refresh_interval"])
+ for cnode_url in cnode_entries:
+ multiaddr = get_ipfs_info_from_cnode_endpoint(cnode_url)
+ update_task.ipfs_client.connect_peer(multiaddr)
+ redis.set(cnode_url, multiaddr, interval)

discovery-provider/src/utils/ipfs_lib.py

@@ -35,7 +35,7 @@ def get_metadata(self, multihash, metadata_format):
  api_metadata = self.get_metadata_from_gateway(multihash, metadata_format)
  retrieved_from_gateway = api_metadata != metadata_format
  except Exception:
- logger.error(
+ logger.info(
  f"Failed to retrieve CID from gateway, {multihash}", exc_info=True
  )
 
@@ -78,7 +78,7 @@ def get_metadata_from_gateway(self, multihash, metadata_format):
  # Skip URL if invalid
  validate_url = urlparse(gateway_query_address)
  if not validate_url.scheme:
- logger.error(
+ logger.info(
  f"IPFSCLIENT | Invalid URL from provided gateway addr - "
  f"provided host: {address} CID address:{gateway_query_address}"
  )
@@ -146,3 +146,13 @@ def cat(self, multihash):
  except:
  logger.error(f"IPFSCLIENT | IPFS cat timed out for CID {multihash}")
  raise # error is of type ipfshttpclient.exceptions.TimeoutError
+
+ def connect_peer(self, peer):
+ try:
+ logger.info('peering with %s', peer)
+ r = self._api.swarm.connect(peer)
+ logger.info(r)
+ return r
+ except Exception as e:
+ logger.error(f"IPFSCLIENT | IPFS Failed to update peer")
+ logger.error(e)

libs/scripts/ipfs.sh

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+
+API_PORT=6001
+SWARM_PORT=6002
+CONTAINER_NAME=local_ipfs_node
+
+if [[ "$1" =~ ^up|down$ ]]; then 
+ echo "Local ipfs operations:"
+else
+ echo "Must be a valid command - ./scripts/ipfs.sh <up|down>"
+ exit 1
+fi
+
+if [[ "$1" == 'up' ]]; then
+ if [[ -z "$2" ]]; then
+ echo "Format - ./scripts/ipfs.sh up <serviceName> <apiPort> <swarmPort>"
+ exit 1
+ else
+ CONTAINER_NAME=$2
+ fi
+
+ if [[ -z "$3" ]]; then
+ echo "Using default apiPort - $API_PORT"
+ echo "Format - ./scripts/ipfs.sh up <serviceName> <apiPort> <swarmPort>"
+ else
+ API_PORT=$3
+ fi
+
+ if [[ -z "$4" ]]; then
+ echo "Using default swarmPort - $SWARM_PORT"
+ echo "Format - ./scripts/ipfs.sh up <serviceName> <apiPort> <swarmPort>"
+ else
+ SWARM_PORT=$4
+ fi
+ docker run -d --name $CONTAINER_NAME -p 127.0.0.1:$API_PORT:5001 -p 127.0.0.1:$SWARM_PORT:4001 ipfs/go-ipfs:release daemon
+elif [[ "$1" == 'down' ]]; then
+ if [[ -z "$2" ]]; then
+ echo "Format - ./scripts/ipfs.sh up <serviceName> <apiPort> <swarmPort>"
+ exit 1
+ else
+ CONTAINER_NAME=$2
+ fi
+ docker stop $CONTAINER_NAME
+ docker rm $CONTAINER_NAME
+fi
+