title: Can Airflow persist access to metadata of short-lived dynamically generated tasks?
body: <p>I have a DAG that, whenever there are files detected by <code>FileSensor</code>, generates tasks for each file to (1) move the file to a staging area, (2) trigger a separate DAG to process the file.</p>
  <pre><code>FileSensor -&gt; Move(File1) -&gt; TriggerDAG(File1) -&gt; Done
            |-&gt; Move(File2) -&gt; TriggerDAG(File2) -^
  </code></pre>
  <p>In the DAG definition file, the middle tasks are generated by iterating over the directory that FileSensor is watching, a bit like this:</p>
  <pre class="lang-py prettyprint-override"><code># def generate_move_task(f: Path) -&gt; BashOperator
  # def generate_dag_trigger(f: Path) -&gt; TriggerDagRunOperator
  
  with dag:
    for filepath in Path(WATCH_DIR).glob(*):
      sensor_task &gt;&gt; generate_move_task(filepath) &gt;&gt; generate_dag_trigger(filepath)
  
  </code></pre>
  <p>The <code>Move</code> task moves the files that lead to the task generation, so the next DAG run won't have <code>FileSensor</code> re-trigger either <code>Move</code> or <code>TriggerDAG</code> tasks for this file. <strong>In fact, the scheduler won't generate the tasks for this file at all, since after all files go through <code>Move</code>, the input directory has no contents to iterate over anymore.</strong>.</p>
  <p>This gives rise to two problems:</p>
  <ol>
  <li><strong>After execution, the task logs and renderings are no longer available.</strong> The Graph View only shows the DAG as it is now (empty), not as it was at runtime. (The Tree View shows that the tasks' run and state, but clicking on the &quot;square&quot; and picking any details leads to an Airflow error.)</li>
  <li><strong>The downstream tasks can be memory-holed due to a race condition.</strong> The first task is to move the originating file to a staging area. If that takes longer than the scheduler polling period, the scheduler no longer collects the downstream <code>TriggerDAG(File1)</code> task, which means that task is not scheduled to be executed even though the upstream task ran successfully. It's as if the downstream task never existed.</li>
  </ol>
  <p>The race condition issue is solved by changing the task sequence to <code>Copy(File1) -&gt; TriggerDAG(File1) -&gt; Remove(File1)</code>, but the broader problem remains: <strong>is there a way to persist dynamically generated tasks, or at least a way to consistently access them through the Airflow interface?</strong></p>