func (d *driver) propagateCommit(stm col.STM, branch *pfs.Branch) error {
	if branch == nil {
		return fmt.Errorf("cannot propagate nil branch")
	}

	// 'subvBranchInfos' is the collection of downstream branches that may get a
	// new commit. Populate subvBranchInfo
	var subvBranchInfos []*pfs.BranchInfo
	branchInfo := &pfs.BranchInfo{}
	if err := d.branches(branch.Repo.Name).ReadWrite(stm).Get(branch.Name, branchInfo); err != nil {
		return err
	}
	subvBranchInfos = append(subvBranchInfos, branchInfo) // add 'branch' itself
	for _, subvBranch := range branchInfo.Subvenance {
		subvBranchInfo := &pfs.BranchInfo{}
		if err := d.branches(subvBranch.Repo.Name).ReadWrite(stm).Get(subvBranch.Name, subvBranchInfo); err != nil {
			return err
		}
		subvBranchInfos = append(subvBranchInfos, subvBranchInfo)
	}

	var head *pfs.CommitInfo
	if branchInfo.Head != nil {
		head = &pfs.CommitInfo{}
		if err := d.commits(branch.Repo.Name).ReadWrite(stm).Get(branchInfo.Head.ID, head); err != nil {
			return err
		}
	}

	// Sort subvBranchInfos so that upstream branches are processed before their
	// descendants. This guarantees that if branch B is provenant on branch A, we
	// create a new commit in A before creating a new commit in B provenant on the
	// (new) HEAD of A.
	sort.Slice(subvBranchInfos, func(i, j int) bool { return len(subvBranchInfos[i].Provenance) < len(subvBranchInfos[j].Provenance) })

	// Iterate through downstream branches and determine which need a new commit.
nextSubvBranch:
	for _, branchInfo := range subvBranchInfos {
		branch := branchInfo.Branch
		repo := branch.Repo
		commits := d.commits(repo.Name).ReadWrite(stm)
		branches := d.branches(repo.Name).ReadWrite(stm)

		// Compute the full provenance of hypothetical new output commit to decide
		// if we need it
		key := path.Join
		commitProvMap := make(map[string]*pfs.CommitProvenance)
		for _, provBranch := range branchInfo.Provenance {
			provBranchInfo := &pfs.BranchInfo{}
			if err := d.branches(provBranch.Repo.Name).ReadWrite(stm).Get(provBranch.Name, provBranchInfo); err != nil && !col.IsErrNotFound(err) {
				return fmt.Errorf("could not read branch %s/%s: %v", provBranch.Repo.Name, provBranch.Name, err)
			}
			if provBranchInfo.Head == nil {
				continue
			}
			// We need to key on both the commit id and the branch name, so that branches with a shared commit are both represented in the provenance
			commitProvMap[key(provBranchInfo.Head.ID, provBranch.Name)] = &pfs.CommitProvenance{
				Commit: provBranchInfo.Head,
				Branch: provBranch,
			}
			// Because provenace is stored as a transitive closure, we don't
			// need to inspect provBranchInfo.HEAD's provenance. Every commit
			// in there will be the HEAD of some other provBranchInfo.
		}
		if head != nil {
			for _, commitProv := range head.Provenance {
				commitProvMap[key(commitProv.Commit.ID, commitProv.Branch.Name)] = commitProv
			}
		}
		if len(commitProvMap) == 0 {
			// no input commits to process; don't create a new output commit
			continue nextSubvBranch
		}

		// 'branch' may already have a HEAD commit, so compute whether the new
		// output commit would have the same provenance as the existing HEAD
		// commit. If so, a new output commit would be a duplicate, so don't create
		// it.
		if branchInfo.Head != nil {
			branchHeadInfo := &pfs.CommitInfo{}
			if err := commits.Get(branchInfo.Head.ID, branchHeadInfo); err != nil {
				return pfsserver.ErrCommitNotFound{branchInfo.Head}
			}
			headIsSubset := false
			for _, v := range commitProvMap {
				matched := false
				for _, c := range branchHeadInfo.Provenance {
					if c.Commit.ID == v.Commit.ID {
						matched = true
					}
				}
				headIsSubset = matched
				if !headIsSubset {
					break
				}
			}
			if len(branchHeadInfo.Provenance) >= len(commitProvMap) && headIsSubset {
				// existing HEAD commit is the same new output commit would be; don't
				// create new commit
				continue nextSubvBranch
			}
		}

		// If the only branches in the hypothetical output commit's provenance are
		// in the 'spec' repo, creating it would mean creating a confusing
		// "dummy" job with no non-spec input data. If this is the case, don't
		// create a new output commit
		allSpec := true
		for _, p := range commitProvMap {
			if p.Branch.Repo.Name != ppsconsts.SpecRepo {
				allSpec = false
				break
			}
		}
		if allSpec {
			// Only input data is PipelineInfo; don't create new output commit
			continue nextSubvBranch
		}

		// *All checks passed* start a new output commit in 'subvBranch'
		newCommit := &pfs.Commit{
			Repo: branch.Repo,
			ID:   uuid.NewWithoutDashes(),
		}
		newCommitInfo := &pfs.CommitInfo{
			Commit:  newCommit,
			Started: now(),
		}

		// Set 'newCommit's ParentCommit, 'branch.Head's ChildCommits and 'branch.Head'
		newCommitInfo.ParentCommit = branchInfo.Head
		if branchInfo.Head != nil {
			parentCommitInfo := &pfs.CommitInfo{}
			if err := commits.Update(newCommitInfo.ParentCommit.ID, parentCommitInfo, func() error {
				parentCommitInfo.ChildCommits = append(parentCommitInfo.ChildCommits, newCommit)
				return nil
			}); err != nil {
				return err
			}
		}
		branchInfo.Head = newCommit
		branchInfo.Name = branch.Name // set in case 'branch' is new
		branchInfo.Branch = branch    // set in case 'branch' is new
		newCommitInfo.Branch = branch
		if err := branches.Put(branch.Name, branchInfo); err != nil {
			return err
		}

		// Set provenance and upstream subvenance (appendSubvenance needs
		// newCommitInfo.ParentCommit to extend the correct subvenance range)
		for _, prov := range commitProvMap {
			// set provenance of 'newCommit'
			newCommitInfo.Provenance = append(newCommitInfo.Provenance, prov)

			// update subvenance of 'prov'
			provCommitInfo := &pfs.CommitInfo{}
			if err := d.commits(prov.Commit.Repo.Name).ReadWrite(stm).Update(prov.Commit.ID, provCommitInfo, func() error {
				appendSubvenance(provCommitInfo, newCommitInfo)
				return nil
			}); err != nil {
				return err
			}
		}

		// finally create open 'commit'
		if err := commits.Create(newCommit.ID, newCommitInfo); err != nil {
			return err
		}
		if err := d.openCommits.ReadWrite(stm).Put(newCommit.ID, newCommit); err != nil {
			return err
		}
	}
	return nil
}
func (d *driver) propagateCommit(stm col.STM, branch *pfs.Branch) error {
	if branch == nil {
		return fmt.Errorf("cannot propagate nil branch")
	}

	
	
	var subvBranchInfos []*pfs.BranchInfo
	branchInfo := &pfs.BranchInfo{}
	if err := d.branches(branch.Repo.Name).ReadWrite(stm).Get(branch.Name, branchInfo); err != nil {
		return err
	}
	subvBranchInfos = append(subvBranchInfos, branchInfo) 
	for _, subvBranch := range branchInfo.Subvenance {
		subvBranchInfo := &pfs.BranchInfo{}
		if err := d.branches(subvBranch.Repo.Name).ReadWrite(stm).Get(subvBranch.Name, subvBranchInfo); err != nil {
			return err
		}
		subvBranchInfos = append(subvBranchInfos, subvBranchInfo)
	}

	var head *pfs.CommitInfo
	if branchInfo.Head != nil {
		head = &pfs.CommitInfo{}
		if err := d.commits(branch.Repo.Name).ReadWrite(stm).Get(branchInfo.Head.ID, head); err != nil {
			return err
		}
	}

	
	
	
	
	sort.Slice(subvBranchInfos, func(i, j int) bool { return len(subvBranchInfos[i].Provenance) < len(subvBranchInfos[j].Provenance) })

	
nextSubvBranch:
	for _, branchInfo := range subvBranchInfos {
		branch := branchInfo.Branch
		repo := branch.Repo
		commits := d.commits(repo.Name).ReadWrite(stm)
		branches := d.branches(repo.Name).ReadWrite(stm)

		
		
		key := path.Join
		commitProvMap := make(map[string]*pfs.CommitProvenance)
		for _, provBranch := range branchInfo.Provenance {
			provBranchInfo := &pfs.BranchInfo{}
			if err := d.branches(provBranch.Repo.Name).ReadWrite(stm).Get(provBranch.Name, provBranchInfo); err != nil && !col.IsErrNotFound(err) {
				return fmt.Errorf("could not read branch %s/%s: %v", provBranch.Repo.Name, provBranch.Name, err)
			}
			if provBranchInfo.Head == nil {
				continue
			}
			
			commitProvMap[key(provBranchInfo.Head.ID, provBranch.Name)] = &pfs.CommitProvenance{
				Commit: provBranchInfo.Head,
				Branch: provBranch,
			}
			
			
			
		}
		if head != nil {
			for _, commitProv := range head.Provenance {
				commitProvMap[key(commitProv.Commit.ID, commitProv.Branch.Name)] = commitProv
			}
		}
		if len(commitProvMap) == 0 {
			
			continue nextSubvBranch
		}

		
		
		
		
		if branchInfo.Head != nil {
			branchHeadInfo := &pfs.CommitInfo{}
			if err := commits.Get(branchInfo.Head.ID, branchHeadInfo); err != nil {
				return pfsserver.ErrCommitNotFound{branchInfo.Head}
			}
			headIsSubset := false
			for _, v := range commitProvMap {
				matched := false
				for _, c := range branchHeadInfo.Provenance {
					if c.Commit.ID == v.Commit.ID {
						matched = true
					}
				}
				headIsSubset = matched
				if !headIsSubset {
					break
				}
			}
			if len(branchHeadInfo.Provenance) >= len(commitProvMap) && headIsSubset {
				
				
				continue nextSubvBranch
			}
		}

		
		
		
		
		allSpec := true
		for _, p := range commitProvMap {
			if p.Branch.Repo.Name != ppsconsts.SpecRepo {
				allSpec = false
				break
			}
		}
		if allSpec {
			
			continue nextSubvBranch
		}

		
		newCommit := &pfs.Commit{
			Repo: branch.Repo,
			ID:   uuid.NewWithoutDashes(),
		}
		newCommitInfo := &pfs.CommitInfo{
			Commit:  newCommit,
			Started: now(),
		}

		
		newCommitInfo.ParentCommit = branchInfo.Head
		if branchInfo.Head != nil {
			parentCommitInfo := &pfs.CommitInfo{}
			if err := commits.Update(newCommitInfo.ParentCommit.ID, parentCommitInfo, func() error {
				parentCommitInfo.ChildCommits = append(parentCommitInfo.ChildCommits, newCommit)
				return nil
			}); err != nil {
				return err
			}
		}
		branchInfo.Head = newCommit
		branchInfo.Name = branch.Name 
		branchInfo.Branch = branch    
		newCommitInfo.Branch = branch
		if err := branches.Put(branch.Name, branchInfo); err != nil {
			return err
		}

		
		
		for _, prov := range commitProvMap {
			
			newCommitInfo.Provenance = append(newCommitInfo.Provenance, prov)

			
			provCommitInfo := &pfs.CommitInfo{}
			if err := d.commits(prov.Commit.Repo.Name).ReadWrite(stm).Update(prov.Commit.ID, provCommitInfo, func() error {
				appendSubvenance(provCommitInfo, newCommitInfo)
				return nil
			}); err != nil {
				return err
			}
		}

		
		if err := commits.Create(newCommit.ID, newCommitInfo); err != nil {
			return err
		}
		if err := d.openCommits.ReadWrite(stm).Put(newCommit.ID, newCommit); err != nil {
			return err
		}
	}
	return nil
}
func ( d * driver ) propagateCommit ( stm col . STM , branch * pfs . Branch ) error { if branch == nil { return fmt . Errorf ( " " ) } // 'subvBranchInfos' is the collection of downstream branches that may get a // new commit. Populate subvBranchInfo var subvBranchInfos [ ] * pfs . BranchInfo branchInfo := & pfs . BranchInfo { } if err := d . branches ( branch . Repo . Name ) . ReadWrite ( stm ) . Get ( branch . Name , branchInfo ) ; err != nil { return err } subvBranchInfos = append ( subvBranchInfos , branchInfo ) // add 'branch' itself for _ , subvBranch := range branchInfo . Subvenance { subvBranchInfo := & pfs . BranchInfo { } if err := d . branches ( subvBranch . Repo . Name ) . ReadWrite ( stm ) . Get ( subvBranch . Name , subvBranchInfo ) ; err != nil { return err } subvBranchInfos = append ( subvBranchInfos , subvBranchInfo ) } var head * pfs . CommitInfo if branchInfo . Head != nil { head = & pfs . CommitInfo { } if err := d . commits ( branch . Repo . Name ) . ReadWrite ( stm ) . Get ( branchInfo . Head . ID , head ) ; err != nil { return err } } // Sort subvBranchInfos so that upstream branches are processed before their // descendants. This guarantees that if branch B is provenant on branch A, we // create a new commit in A before creating a new commit in B provenant on the // (new) HEAD of A. sort . Slice ( subvBranchInfos , func ( i , j int ) bool { return len ( subvBranchInfos [ i ] . Provenance ) < len ( subvBranchInfos [ j ] . Provenance ) } ) // Iterate through downstream branches and determine which need a new commit. nextSubvBranch : for _ , branchInfo := range subvBranchInfos { branch := branchInfo . Branch repo := branch . Repo commits := d . commits ( repo . Name ) . ReadWrite ( stm ) branches := d . branches ( repo . Name ) . ReadWrite ( stm ) // Compute the full provenance of hypothetical new output commit to decide // if we need it key := path . Join commitProvMap := make ( map [ string ] * pfs . CommitProvenance ) for _ , provBranch := range branchInfo . Provenance { provBranchInfo := & pfs . BranchInfo { } if err := d . branches ( provBranch . Repo . Name ) . ReadWrite ( stm ) . Get ( provBranch . Name , provBranchInfo ) ; err != nil && ! col . IsErrNotFound ( err ) { return fmt . Errorf ( " " , provBranch . Repo . Name , provBranch . Name , err ) } if provBranchInfo . Head == nil { continue } // We need to key on both the commit id and the branch name, so that branches with a shared commit are both represented in the provenance commitProvMap [ key ( provBranchInfo . Head . ID , provBranch . Name ) ] = & pfs . CommitProvenance { Commit : provBranchInfo . Head , Branch : provBranch , } // Because provenace is stored as a transitive closure, we don't // need to inspect provBranchInfo.HEAD's provenance. Every commit // in there will be the HEAD of some other provBranchInfo. } if head != nil { for _ , commitProv := range head . Provenance { commitProvMap [ key ( commitProv . Commit . ID , commitProv . Branch . Name ) ] = commitProv } } if len ( commitProvMap ) == 0 { // no input commits to process; don't create a new output commit continue nextSubvBranch } // 'branch' may already have a HEAD commit, so compute whether the new // output commit would have the same provenance as the existing HEAD // commit. If so, a new output commit would be a duplicate, so don't create // it. if branchInfo . Head != nil { branchHeadInfo := & pfs . CommitInfo { } if err := commits . Get ( branchInfo . Head . ID , branchHeadInfo ) ; err != nil { return pfsserver . ErrCommitNotFound { branchInfo . Head } } headIsSubset := false for _ , v := range commitProvMap { matched := false for _ , c := range branchHeadInfo . Provenance { if c . Commit . ID == v . Commit . ID { matched = true } } headIsSubset = matched if ! headIsSubset { break } } if len ( branchHeadInfo . Provenance ) >= len ( commitProvMap ) && headIsSubset { // existing HEAD commit is the same new output commit would be; don't // create new commit continue nextSubvBranch } } // If the only branches in the hypothetical output commit's provenance are // in the 'spec' repo, creating it would mean creating a confusing // "dummy" job with no non-spec input data. If this is the case, don't // create a new output commit allSpec := true for _ , p := range commitProvMap { if p . Branch . Repo . Name != ppsconsts . SpecRepo { allSpec = false break } } if allSpec { // Only input data is PipelineInfo; don't create new output commit continue nextSubvBranch } // *All checks passed* start a new output commit in 'subvBranch' newCommit := & pfs . Commit { Repo : branch . Repo , ID : uuid . NewWithoutDashes ( ) , } newCommitInfo := & pfs . CommitInfo { Commit : newCommit , Started : now ( ) , } // Set 'newCommit's ParentCommit, 'branch.Head's ChildCommits and 'branch.Head' newCommitInfo . ParentCommit = branchInfo . Head if branchInfo . Head != nil { parentCommitInfo := & pfs . CommitInfo { } if err := commits . Update ( newCommitInfo . ParentCommit . ID , parentCommitInfo , func ( ) error { parentCommitInfo . ChildCommits = append ( parentCommitInfo . ChildCommits , newCommit ) return nil } ) ; err != nil { return err } } branchInfo . Head = newCommit branchInfo . Name = branch . Name // set in case 'branch' is new branchInfo . Branch = branch // set in case 'branch' is new newCommitInfo . Branch = branch if err := branches . Put ( branch . Name , branchInfo ) ; err != nil { return err } // Set provenance and upstream subvenance (appendSubvenance needs // newCommitInfo.ParentCommit to extend the correct subvenance range) for _ , prov := range commitProvMap { // set provenance of 'newCommit' newCommitInfo . Provenance = append ( newCommitInfo . Provenance , prov ) // update subvenance of 'prov' provCommitInfo := & pfs . CommitInfo { } if err := d . commits ( prov . Commit . Repo . Name ) . ReadWrite ( stm ) . Update ( prov . Commit . ID , provCommitInfo , func ( ) error { appendSubvenance ( provCommitInfo , newCommitInfo ) return nil } ) ; err != nil { return err } } // finally create open 'commit' if err := commits . Create ( newCommit . ID , newCommitInfo ) ; err != nil { return err } if err := d . openCommits . ReadWrite ( stm ) . Put ( newCommit . ID , newCommit ) ; err != nil { return err } } return nil }
// propagateCommit selectively starts commits in or downstream of 'branch' in
// order to restore the invariant that branch provenance matches HEAD commit
// provenance:
//   B.Head is provenant on A.Head <=>
//   branch B is provenant on branch A and A.Head != nil
// The implementation assumes that the invariant already holds for all branches
// upstream of 'branch', but not necessarily for 'branch' itself. Despite the
// name, 'branch' does not need a HEAD commit to propagate, though one may be
// created.
//
// In other words, propagateCommit scans all branches b_downstream that are
// equal to or downstream of 'branch', and if the HEAD of b_downstream isn't
// provenant on the HEADs of b_downstream's provenance, propagateCommit starts
// a new HEAD commit in b_downstream that is. For example, propagateCommit
// starts downstream output commits (which trigger PPS jobs) when new input
// commits arrive on 'branch', when 'branch's HEAD is deleted, or when 'branch'
// is newly created (i.e. in CreatePipeline).
propagateCommit selectively starts commits in or downstream of branch in order to restore the invariant that branch provenance matches HEAD commit provenance : B . Head is provenant on A . Head < = > branch B is provenant on branch A and A . Head ! = nil The implementation assumes that the invariant already holds for all branches upstream of branch but not necessarily for branch itself . Despite the name branch does not need a HEAD commit to propagate though one may be created . In other words propagateCommit scans all branches b_downstream that are equal to or downstream of branch and if the HEAD of b_downstream isn t provenant on the HEADs of b_downstream s provenance propagateCommit starts a new HEAD commit in b_downstream that is . For example propagateCommit starts downstream output commits ( which trigger PPS jobs ) when new input commits arrive on branch when branch s HEAD is deleted or when branch is newly created ( i . e . in CreatePipeline ) .