Function recomputeNodeSelfOnly

pj_datastore/src/derived_engine.cpp:1234–1275

Clear ONE node's output, reset its op + watermarks, and fully replay its input. The per-node primitive behind recomputeBatch's self+downstream cascade — does NOT touch downstream nodes. clearChunks (not retireTopic) keeps every TopicStorage alive so cached reader pointers stay valid.

// downstream nodes. clearChunks (not retireTopic) keeps every TopicStorage alive so cached
// reader pointers stay valid.
static PJ::Status recomputeNodeSelfOnly(DerivedEngineImpl& impl, DataEngine& engine, DerivedNode& node) {
  // 1. Clear all output chunks unconditionally.
  for (PJ::TopicId out_tid : node.output_topic_ids) {
    TopicStorage* storage = engine.getTopicStorage(out_tid);
    if (storage) {
      storage->clearChunks();
    }
  }

  // 2. Reset transform state
  if (!node.is_mimo) {
    if (node.siso_op) {
      node.siso_op->reset();
    }
  } else {
    if (node.mimo_op) {
      node.mimo_op->reset();
    }
  }

  // 3. Reset processed watermarks (chunk ids and timestamps)
  node.last_processed_chunk_id = 0;
  node.siso_last_ts = std::numeric_limits<PJ::Timestamp>::min();
  if (node.is_mimo) {
    node.mimo_last_ts = std::numeric_limits<PJ::Timestamp>::min();
    node.mimo_last_chunk_id = 0;
  }

  // 4. Full replay
  PJ::Status s = PJ::okStatus();
  if (!node.is_mimo) {
    s = runSisoIncremental(impl, engine, node);
  } else {
    s = runMimoIncremental(impl, engine, node);
  }

  if (!s.has_value()) {
    return s;
  }
  node.dirty = false;
  return PJ::okStatus();
}

PJ::Status DerivedEngine::recomputeBatch(PJ::NodeId node_id) {
  auto lock = engine_.lockEngine();  // exclusive for the whole clear+replay cascade