Hi everyone, I have been playing around with the "run-until" code github.com/nextflow-io/run-until from Robert Syme's recent blog post for iterative workflows (seqera.io/blog/running-until-iterative-loops-in-nextflow), and I ran into a race condition I can't figure out. Here's what I'm seeing - with the topic channel +

until

pattern, longer-running tasks seems to trigger a race condition.

max_iter

is user configurable and defaults to 2, but the pipeline consistently submits extra iterations like 3-4 and more instead of stopping at 2 as expected. Interestingly, it works fine in nf-test, which serializes execution more tightly, but overshoots when I run it via nextflow run. I have event tested my logic on the run-until code by changing the limit it works perfectly! My setup is nextflow 26.04.6 on macOS, using the local executor with a conda profile. Each iteration takes about ~3-5 minutes since it involves longer-running compute tasks. Here's a minimal version of my code

workflow ITER_WF {
  main:
  ch_input1 = channel.fromPath(params.input1)
  ch_input2 = channel.fromPath(params.input2)
  ch_input3 = channel.fromPath(params.input3)
  SETUP(ch_input1, ch_input2, ch_input3)

  def parser = new groovy.json.JsonSlurper()
  updates = channel.topic('iteration_state')
  limit = updates.until { path -> parser.parseText(path.text).step >= params.max_iter }
  ITERATE(updates, limit)
}

the process code -

process SETUP {
  input:
  path input1
  path input2
  path input3

  output:
  path "state.json", topic: 'iteration_state'

  script:
  """
  python3 -c "
import json, os
state = {'step': 0, 'input1': os.path.abspath('${input1}'), 'input2': os.path.abspath('${input2}'), 'input3': os.path.abspath('${input3}')}
with open('state.json', 'w') as f:
    json.dump(state, f)
"
  """
}

process ITERATE {
  input:
  path state_json
  val limiter

  output:
  path "state.json", topic: 'iteration_state'

  script:
  """
  #!/usr/bin/env python3
  import json

  with open('${state_json}') as f:
      state = json.load(f)

  step = state['step'] + 1
  # ... a few minutes of compute happens here ...
  new_state = {'step': step}
  with open('state.json', 'w') as f:
      json.dump(new_state, f)
  """
}

I have tried using

max_iter  - 1

in

until

condition as a workaround, but that didn't fix it either. I am guessing that with two-input pattern (

updates

+

limit

), there's a race where

updates

delivers a new item to the process before

limit

evaluates that same item and closes things out; while slower tasks just give that race more room to happen. Curious if anyone has run into something similar, or has thoughts on what might fix it? Thanks! cat eyes

This is not a race condition. Rather, you're running into type confusion because of

params

not being autocasted into some other type (but I guess you working nf-test example provides params as a json or yaml instead).

Thank you @pontus, this makes a lot of sense. Honestly, didn't think of it like this at all, was so focused on race condition that I missed the obvious type issue. Added an explicit cast to

Integer

for

max_iter

in the

until

closure, and that did the trick!

I suspect you may need to ask that more visibly, possibly rather in the nextflow slack.