|
|
@@ -19,83 +19,115 @@
|
|
|
'use strict'
|
|
|
|
|
|
const debug = require('debug')('joystream:sync')
|
|
|
+const _ = require('lodash')
|
|
|
|
|
|
-async function syncCallback(api, storage) {
|
|
|
- // The first step is to gather all data objects from chain.
|
|
|
- // TODO: in future, limit to a configured tranche
|
|
|
- // FIXME this isn't actually on chain yet, so we'll fake it.
|
|
|
- const knownContentIds = (await api.assets.getKnownContentIds()) || []
|
|
|
+// The number of concurrent sync sessions allowed. Must be greater than zero.
|
|
|
+const MAX_CONCURRENT_SYNC_ITEMS = 20
|
|
|
|
|
|
- const roleAddress = api.identities.key.address
|
|
|
- const providerId = api.storageProviderId
|
|
|
+async function syncContent({ api, storage, contentBeingSynced, contentCompleteSynced }) {
|
|
|
+ const knownContentIds = await api.assets.getKnownContentIds()
|
|
|
|
|
|
- // Iterate over all sync objects, and ensure they're synced.
|
|
|
- const allChecks = knownContentIds.map(async (contentId) => {
|
|
|
- // eslint-disable-next-line prefer-const
|
|
|
- let { relationship, relationshipId } = await api.assets.getStorageRelationshipAndId(providerId, contentId)
|
|
|
+ // Select ids which we have not yet fully synced
|
|
|
+ const needsSync = knownContentIds
|
|
|
+ .filter((id) => !contentCompleteSynced.has(id))
|
|
|
+ .filter((id) => !contentBeingSynced.has(id))
|
|
|
|
|
|
- // get the data object
|
|
|
- // make sure the data object was Accepted by the liaison,
|
|
|
- // don't just blindly attempt to fetch them
|
|
|
+ // Since we are limiting concurrent content ids being synced, to ensure
|
|
|
+ // better distribution of content across storage nodes during a potentially long
|
|
|
+ // sync process we don't want all nodes to replicate items in the same order, so
|
|
|
+ // we simply shuffle.
|
|
|
+ const candidatesForSync = _.shuffle(needsSync)
|
|
|
+
|
|
|
+ // TODO: get the data object
|
|
|
+ // make sure the data object was Accepted by the liaison,
|
|
|
+ // don't just blindly attempt to fetch them
|
|
|
+ while (contentBeingSynced.size < MAX_CONCURRENT_SYNC_ITEMS && candidatesForSync.length) {
|
|
|
+ const contentId = candidatesForSync.shift()
|
|
|
|
|
|
- let fileLocal
|
|
|
try {
|
|
|
- // check if we have content or not
|
|
|
- const stats = await storage.stat(contentId)
|
|
|
- fileLocal = stats.local
|
|
|
+ contentBeingSynced.set(contentId)
|
|
|
+ await storage.synchronize(contentId, (err, status) => {
|
|
|
+ if (err) {
|
|
|
+ contentBeingSynced.delete(contentId)
|
|
|
+ debug(`Error Syncing ${err}`)
|
|
|
+ } else if (status.synced) {
|
|
|
+ contentBeingSynced.delete(contentId)
|
|
|
+ contentCompleteSynced.set(contentId)
|
|
|
+ }
|
|
|
+ })
|
|
|
} catch (err) {
|
|
|
- // on error stating or timeout
|
|
|
- debug(err.message)
|
|
|
- // we don't have content if we can't stat it
|
|
|
- fileLocal = false
|
|
|
+ // Most likely failed to resolve the content id
|
|
|
+ debug(`Failed calling synchronize ${err}`)
|
|
|
+ contentBeingSynced.delete(contentId)
|
|
|
}
|
|
|
+ }
|
|
|
+}
|
|
|
|
|
|
- if (!fileLocal) {
|
|
|
- try {
|
|
|
- await storage.synchronize(contentId)
|
|
|
- } catch (err) {
|
|
|
- // duplicate logging
|
|
|
- // debug(err.message)
|
|
|
- return
|
|
|
- }
|
|
|
- // why are we returning, if we synced the file
|
|
|
- return
|
|
|
- }
|
|
|
+async function createNewRelationships({ api, contentCompleteSynced }) {
|
|
|
+ const roleAddress = api.identities.key.address
|
|
|
+ const providerId = api.storageProviderId
|
|
|
|
|
|
- if (!relationship) {
|
|
|
- // create relationship
|
|
|
- debug(`Creating new storage relationship for ${contentId.encode()}`)
|
|
|
- try {
|
|
|
- relationshipId = await api.assets.createStorageRelationship(roleAddress, providerId, contentId)
|
|
|
- await api.assets.toggleStorageRelationshipReady(roleAddress, providerId, relationshipId, true)
|
|
|
- } catch (err) {
|
|
|
- debug(`Error creating new storage relationship ${contentId.encode()}: ${err.stack}`)
|
|
|
- }
|
|
|
- } else if (!relationship.ready) {
|
|
|
- debug(`Updating storage relationship to ready for ${contentId.encode()}`)
|
|
|
- // update to ready. (Why would there be a relationship set to ready: false?)
|
|
|
+ // Create new relationships for synced content if required and
|
|
|
+ // compose list of relationship ids to be set to ready.
|
|
|
+ return (
|
|
|
+ await Promise.all(
|
|
|
+ [...contentCompleteSynced.keys()].map(async (contentId) => {
|
|
|
+ const { relationship, relationshipId } = await api.assets.getStorageRelationshipAndId(providerId, contentId)
|
|
|
+
|
|
|
+ if (relationship) {
|
|
|
+ // maybe prior transaction to set ready failed for some reason..
|
|
|
+ if (!relationship.ready) {
|
|
|
+ return relationshipId
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ // create relationship
|
|
|
+ debug(`Creating new storage relationship for ${contentId.encode()}`)
|
|
|
+ try {
|
|
|
+ return await api.assets.createStorageRelationship(roleAddress, providerId, contentId)
|
|
|
+ } catch (err) {
|
|
|
+ debug(`Error creating new storage relationship ${contentId.encode()}: ${err.stack}`)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return null
|
|
|
+ })
|
|
|
+ )
|
|
|
+ ).filter((id) => id !== null)
|
|
|
+}
|
|
|
+
|
|
|
+async function setRelationshipsReady({ api, relationshipIds }) {
|
|
|
+ const roleAddress = api.identities.key.address
|
|
|
+ const providerId = api.storageProviderId
|
|
|
+
|
|
|
+ return Promise.all(
|
|
|
+ relationshipIds.map(async (relationshipId) => {
|
|
|
try {
|
|
|
await api.assets.toggleStorageRelationshipReady(roleAddress, providerId, relationshipId, true)
|
|
|
} catch (err) {
|
|
|
- debug(`Error setting relationship ready ${contentId.encode()}: ${err.stack}`)
|
|
|
+ debug('Error setting relationship ready')
|
|
|
}
|
|
|
- } else {
|
|
|
- // we already have content and a ready relationship set. No need to do anything
|
|
|
- // debug(`content already stored locally ${contentId.encode()}`);
|
|
|
- }
|
|
|
- })
|
|
|
-
|
|
|
- return Promise.all(allChecks)
|
|
|
+ })
|
|
|
+ )
|
|
|
}
|
|
|
|
|
|
-async function syncPeriodic(api, flags, storage) {
|
|
|
+async function syncPeriodic({ api, flags, storage, contentBeingSynced, contentCompleteSynced }) {
|
|
|
+ const retry = () => {
|
|
|
+ setTimeout(syncPeriodic, flags.syncPeriod, {
|
|
|
+ api,
|
|
|
+ flags,
|
|
|
+ storage,
|
|
|
+ contentBeingSynced,
|
|
|
+ contentCompleteSynced,
|
|
|
+ })
|
|
|
+ }
|
|
|
+
|
|
|
try {
|
|
|
debug('Starting sync run...')
|
|
|
|
|
|
const chainIsSyncing = await api.chainIsSyncing()
|
|
|
if (chainIsSyncing) {
|
|
|
debug('Chain is syncing. Postponing sync run.')
|
|
|
- return setTimeout(syncPeriodic, flags.syncPeriod, api, flags, storage)
|
|
|
+ return retry()
|
|
|
}
|
|
|
|
|
|
const recommendedBalance = await api.providerHasMinimumBalance(300)
|
|
|
@@ -106,20 +138,26 @@ async function syncPeriodic(api, flags, storage) {
|
|
|
const sufficientBalance = await api.providerHasMinimumBalance(100)
|
|
|
if (!sufficientBalance) {
|
|
|
debug('Provider role account does not have sufficient balance. Postponing sync run!')
|
|
|
- return setTimeout(syncPeriodic, flags.syncPeriod, api, flags, storage)
|
|
|
+ return retry()
|
|
|
}
|
|
|
|
|
|
- await syncCallback(api, storage)
|
|
|
+ await syncContent({ api, storage, contentBeingSynced, contentCompleteSynced })
|
|
|
+ const relationshipIds = await createNewRelationships({ api, contentCompleteSynced })
|
|
|
+ await setRelationshipsReady({ api, relationshipIds })
|
|
|
+
|
|
|
debug('sync run complete')
|
|
|
} catch (err) {
|
|
|
- debug(`Error in syncPeriodic ${err.stack}`)
|
|
|
+ debug(`Error in sync run ${err.stack}`)
|
|
|
}
|
|
|
+
|
|
|
// always try again
|
|
|
- setTimeout(syncPeriodic, flags.syncPeriod, api, flags, storage)
|
|
|
+ retry()
|
|
|
}
|
|
|
|
|
|
function startSyncing(api, flags, storage) {
|
|
|
- syncPeriodic(api, flags, storage)
|
|
|
+ const contentBeingSynced = new Map()
|
|
|
+ const contentCompleteSynced = new Map()
|
|
|
+ syncPeriodic({ api, flags, storage, contentBeingSynced, contentCompleteSynced })
|
|
|
}
|
|
|
|
|
|
module.exports = {
|
Martin