chore: extract deal observer loop

[NikolasHaimerl]

This PR introduces the following changes:

1. Extract the logic of the deal observer loop into its own file
2. Add a test for the deal observer loop
3. Extract the mocked rpc endpoint into its own file for easier reuse.

[bajtos]

Great idea to move the loop functions from bin/ to lib/!

[juliangruber]

This method has a lot of arguments, working with positional arguments is hard in this case. Imagine using an editor that doesn't infer types, or if we later on change the arguments list.

What do you think about changing this to an options object?

[bajtos]

+1 to use an options object (named parameters).

[NikolasHaimerl]

@bajtos isn't this in contrast to what you mentioned in an earlier PR?

[juliangruber]

In the earlier PR @bajtos also suggested to use an options object / param object / named parameters

[NikolasHaimerl]

I see, I must have misunderstood then.

[bajtos]

@bajtos isn't this in contrast to what you mentioned in an earlier PR?

It's great that you asked. As Julian pointed out, I meant the same thing in that other comment.

JavaScript language does not provide syntax for named parameters. We implement named parameters by passing around an object where each property corresponds to one parameter.

// definition
function doSomeStuff({ pgClient, signal }) {
  // ...
}

// example usage
const pgClient = /*..*/

doSomeStuff({
  pgClient,
  signal: AbortSignal.timeout(10_000)
})

[bajtos]

The TypeScript typings are slightly more complex - you need to describe the single positional parameter accepting an object, and then describe the properties of that object.

/**
 * @param {object} params
 * @param {pg.Client} params.pgClient
 * @param {AbortSignal} signal
 */
function doSomeStuff({ pgClient, signal }) {
  // ...
}

I personally tend to call the object argument as args, see e.g. here:

https://github.com/CheckerNetwork/piece-indexer/blob/765dbfe7705ce375f7b5a317f4912264e7a70ffa/indexer/lib/ipni-watcher.js#L11-L16

/**
 * @param {object} args
 * @param {number} args.minSyncIntervalInMs
 * @param {AbortSignal} [args.signal]
 */
export async function * runIpniSync ({ minSyncIntervalInMs, signal }) {
  // ...
}

The syntax [args.signal] tells the type-checker that signal is an optional parameter.

[NikolasHaimerl]

I am not sure I quite understand. How does
({minSyncIntervalInMs, signal}) differ from (minSyncIntervalInMs, signal) for readability. Or do you mean to have a single argument (args) and then call args.pgPool?

[juliangruber]

1. Depending on the calling situation, you might not have named arguments. So, the second one quickly becomes (1000, signal) vs the first one is always clear: ({ minSyncIntervalInMs: 1000, signal })
2. Arguments order doesn't matter with an object, ({ a, b }) works just like ({ b, a })

There are probably more factors here, but these are the ones most important in my mind

[juliangruber]

I'm not sure this will scale well. @bajtos what do you think about this helper? One test might need a different test data set being returned than another. My suspicion is that it will scale better to define this makeRpcRequest function where the tests are also defined (and potentially doing it multiple times).

[bajtos]

I share your concerns and agree it's best to define the stub makerRpcRequest function in every test.

We can implement a builder helper to make it easier to implement such stubs.

// usage in tests
const makeRpcRequest = buildMakeRpcRequest({
  'Filecoin.ChainHead': chainHeadTestData
})

// implementation
export const buildMakeRpcRequest = (methodsToResults) => {
  return async (method, _params) => {
    if (methodsToResults[method]) return methodsToResults[method]
    throw new Error(`Unsupported RPC API method: "${method}"`)
  }
}

[NikolasHaimerl]

Isn"t this adding quite a bit of complexity for a simple problem?
We do not need this complexity right now, and it is also not foreseeable whether we will need it in the future. I would much rather add complexity when we need it instead of before.

[juliangruber]

Wouldn't it be less complex to inline this function (without the helper), than creating a new utility (i.e. a new abstraction)

[bajtos]

+1 to use an options object (named parameters).

[bajtos]

Would you mind not cleaning the DB after tests in this pull request and waiting until we resolve the discussion in #41?

[bajtos]

Nitpick - feel free to ignore.

Fetching data for all rows is not a very efficient way to count the number of rows in a table. It does not matter in this test, since you are fetching up to 360 rows, but it could become a problem if used elsewhere.

A more efficient solution is to use the SQL aggregate function COUNT().

while (true) {
  const { rows } = await pgPool.query('SELECT COUNT(*) FROM active_deals')
  if (rows[0].count === 360) break
}

[bajtos]

I share your concerns and agree it's best to define the stub makerRpcRequest function in every test.

We can implement a builder helper to make it easier to implement such stubs.

// usage in tests
const makeRpcRequest = buildMakeRpcRequest({
  'Filecoin.ChainHead': chainHeadTestData
})

// implementation
export const buildMakeRpcRequest = (methodsToResults) => {
  return async (method, _params) => {
    if (methodsToResults[method]) return methodsToResults[method]
    throw new Error(`Unsupported RPC API method: "${method}"`)
  }
}

[pyropy]

Do you think that extracting the loop logic to a separate function could be a good idea? I had something like this in mind:

// lib/loop.js
const loop = async ({ signal, name, interval, fn, recordTelemetry, captureException }) => {
  const submitDeals = submitDealsToSparkApi(sparkApiBaseURL, dealIngestionAccessToken)
  while (!signal?.aborted) {
    const start = Date.now()
    try {
      await fn()
    } catch (e) {
      console.error(e)
      captureException(e)
    }
    const dt = Date.now() - start
    console.log(`Loop "${name}" took ${dt}ms`)
    recordTelemetry(`loop_${slug(name, '_')}`, point => {
      point.intField('interval_ms', name)
      point.intField('duration_ms', dt)
    })

    if (dt < interval) {
      await timers.setTimeout(interval - dt)
    }
  }
}

// lib/deal-observer-loop.js
const dealObserverLoop = (pgPool, rpcClient, opts) => async () => {
  // logic goes here
}

// lib/deal-submitter-loop.js
const dealSubmitLoop = (pgPool, opts) => async () => {
  // logic goes here
}

// bin/deal-observer-backend.js
await Promise.all([
  loop({
    name: 'dealObserver',
    interval: 1000,
    fn: dealObserverLoop(pgPool, rpcClient, opts),
    recordTelemetry: console.log,
    captureException: console.error,
  }),
  loop({
    name: 'dealSubmit',
    interval: 1000,
    fn: dealSubmitLoop(pgPool, opts),
    recordTelemetry: console.log,
    captureException: console.error,
  })
])

I feel that something like this could help us to keep code more DRY, while also keeping the loop function itself clean as we don't have to pass so many arguments around (sentry, influx, and other dependencies).

[NikolasHaimerl]

This is also a valid approach. Given that all loops follow the same overall structure outside of await fn()

[pyropy]

@bajtos @juliangruber I'd appreciate your input on this as I'd like to apply changes to #33 as well

[juliangruber]

As discussed, #33 can be landed without the loop refactor (can be followed up with)