int pmixp_coll_tree_parent(pmixp_coll_t *coll, uint32_t peerid,
			   uint32_t seq, Buf buf)
{
	pmixp_coll_tree_t *tree = NULL;
	char *data_src = NULL, *data_dst = NULL;
	uint32_t size;
	int expected_peerid;

	/* lock the structure */
	slurm_mutex_lock(&coll->lock);
	tree = &coll->state.tree;

	if (pmixp_info_srv_direct_conn()) {
		expected_peerid = tree->prnt_peerid;
	} else {
		expected_peerid = tree->root_peerid;
	}

	if (expected_peerid != peerid) {
		char *nodename = pmixp_info_job_host(peerid);
		/* protect ourselfs if we are running with no asserts */
		PMIXP_ERROR("%p: parent contrib from bad nodeid=%s:%u, expect=%d",
			    coll, nodename, peerid, expected_peerid);
		xfree(nodename);
		goto proceed;
	}

#ifdef PMIXP_COLL_DEBUG
	PMIXP_DEBUG("%p: contrib/rem nodeid=%u: state=%s, size=%u",
		    coll, peerid, pmixp_coll_tree_state2str(tree->state),
		    remaining_buf(buf));
#endif

	switch (tree->state) {
	case PMIXP_COLL_TREE_SYNC:
	case PMIXP_COLL_TREE_COLLECT:
		/* It looks like a retransmission attempt when remote side
		 * identified transmission failure, but we actually successfuly
		 * received the message */
#ifdef PMIXP_COLL_DEBUG
		PMIXP_DEBUG("%p: prev contrib nodeid=%u: seq=%u, cur_seq=%u, state=%s",
			    coll, peerid, seq, coll->seq,
			    pmixp_coll_tree_state2str(tree->state));
#endif
		/* sanity check */
		if ((coll->seq - 1) != seq) {
			/* FATAL: should not happen in normal workflow */
			char *nodename = pmixp_info_job_host(peerid);
			PMIXP_ERROR("%p: unexpected from %s:%d: seq = %d, coll->seq = %d, state=%s",
				    coll, nodename, peerid, seq, coll->seq,
				    pmixp_coll_tree_state2str(tree->state));
			xfree(nodename);
			goto error;
		}
		goto proceed;
	case PMIXP_COLL_TREE_UPFWD_WSC:{
		/* we are not actually ready to receive this contribution as
		 * the upward portion of the collective wasn't received yet.
		 * This should not happen as SAPI (Slurm API) is blocking and
		 * we chould transit to PMIXP_COLL_UPFWD_WPC immediately */
		/* FATAL: should not happen in normal workflow */
		char *nodename = pmixp_info_job_host(peerid);
		PMIXP_ERROR("%p: unexpected from %s:%d: seq = %d, coll->seq = %d, state=%s",
			    coll, nodename, peerid, seq, coll->seq,
			    pmixp_coll_tree_state2str(tree->state));
		xfree(nodename);
		goto error;
	}
	case PMIXP_COLL_TREE_UPFWD:
	case PMIXP_COLL_TREE_UPFWD_WPC:
		/* we were waiting for this */
		break;
	case PMIXP_COLL_TREE_DOWNFWD:
		/* It looks like a retransmission attempt when remote side
		 * identified transmission failure, but we actually successfuly
		 * received the message */
#ifdef PMIXP_COLL_DEBUG
		PMIXP_DEBUG("%p: double contrib nodeid=%u seq=%u, cur_seq=%u, state=%s",
			    coll, peerid, seq, coll->seq,
			    pmixp_coll_tree_state2str(tree->state));
#endif
		/* sanity check */
		if (coll->seq != seq) {
			char *nodename = pmixp_info_job_host(peerid);
			/* FATAL: should not happen in normal workflow */
			PMIXP_ERROR("%p: unexpected contrib from %s:%d: seq = %d, coll->seq = %d, state=%s",
				    coll, nodename, peerid, seq, coll->seq,
				    pmixp_coll_tree_state2str(tree->state));
			xfree(nodename);
			goto error;
		}
		goto proceed;
	default:
		/* should not happen in normal workflow */
		PMIXP_ERROR("%p: unknown collective state %s",
			    coll, pmixp_coll_tree_state2str(tree->state));
		/* collective is spoiled, reset state */
		tree->state = PMIXP_COLL_TREE_SYNC;
		goto error2;
	}

	/* Because of possible timeouts/delays in transmission we
	 * can receive a contribution second time. Avoid duplications
	 * by checking our records. */
	if (tree->contrib_prnt) {
		char *nodename = pmixp_info_job_host(peerid);
		/* May be 0 or 1. If grater - transmission skew, ignore.
		 * NOTE: this output is not on the critical path -
		 * don't preprocess it out */
		PMIXP_DEBUG("%p: multiple contributions from parent %s:%d",
			    coll, nodename, peerid);
		xfree(nodename);
		/* this is duplication, skip. */
		goto proceed;
	}
	tree->contrib_prnt = true;

	data_src = get_buf_data(buf) + get_buf_offset(buf);
	size = remaining_buf(buf);
	pmixp_server_buf_reserve(tree->dfwd_buf, size);

	data_dst = get_buf_data(tree->dfwd_buf) +
		get_buf_offset(tree->dfwd_buf);
	memcpy(data_dst, data_src, size);
	set_buf_offset(tree->dfwd_buf,
		       get_buf_offset(tree->dfwd_buf) + size);
proceed:
	_progress_coll_tree(coll);

#ifdef PMIXP_COLL_DEBUG
	PMIXP_DEBUG("%p: finish: nodeid=%u, state=%s",
		    coll, peerid, pmixp_coll_tree_state2str(tree->state));
#endif
	/* unlock the structure */
	slurm_mutex_unlock(&coll->lock);

	return SLURM_SUCCESS;
error:
	pmixp_coll_log(coll);
	_reset_coll(coll);
error2:
	slurm_kill_job_step(pmixp_info_jobid(),
			    pmixp_info_stepid(), SIGKILL);
	slurm_mutex_unlock(&coll->lock);

	return SLURM_ERROR;
}