func (d *driver) createBranch(pachClient *client.APIClient, branch *pfs.Branch, commit *pfs.Commit, provenance []*pfs.Branch) error {
	ctx := pachClient.Ctx()
	if err := d.checkIsAuthorized(pachClient, branch.Repo, auth.Scope_WRITER); err != nil {
		return err
	}
	// Validate request. The request must do exactly one of:
	// 1) updating 'branch's provenance (commit is nil OR commit == branch)
	// 2) re-pointing 'branch' at a new commit
	if commit != nil {
		// Determine if this is a provenance update
		sameTarget := branch.Repo.Name == commit.Repo.Name && branch.Name == commit.ID
		if !sameTarget && provenance != nil {
			return fmt.Errorf("cannot point branch \"%s\" at target commit \"%s/%s\" without clearing its provenance",
				branch.Name, commit.Repo.Name, commit.ID)
		}
	}

	_, err := col.NewSTM(ctx, d.etcdClient, func(stm col.STM) error {
		// if 'commit' is a branch, resolve it
		var err error
		if commit != nil {
			_, err = d.resolveCommit(stm, commit) // if 'commit' is a branch, resolve it
			if err != nil {
				// possible that branch exists but has no head commit. This is fine, but
				// branchInfo.Head must also be nil
				if !isNoHeadErr(err) {
					return fmt.Errorf("unable to inspect %s/%s: %v", err, commit.Repo.Name, commit.ID)
				}
				commit = nil
			}
		}

		// Retrieve (and create, if necessary) the current version of this branch
		branches := d.branches(branch.Repo.Name).ReadWrite(stm)
		branchInfo := &pfs.BranchInfo{}
		if err := branches.Upsert(branch.Name, branchInfo, func() error {
			branchInfo.Name = branch.Name // set in case 'branch' is new
			branchInfo.Branch = branch
			branchInfo.Head = commit
			branchInfo.DirectProvenance = nil
			for _, provBranch := range provenance {
				add(&branchInfo.DirectProvenance, provBranch)
			}
			return nil
		}); err != nil {
			return err
		}
		repos := d.repos.ReadWrite(stm)
		repoInfo := &pfs.RepoInfo{}
		if err := repos.Update(branch.Repo.Name, repoInfo, func() error {
			add(&repoInfo.Branches, branch)
			return nil
		}); err != nil {
			return err
		}

		// Update (or create)
		// 1) 'branch's Provenance
		// 2) the Provenance of all branches in 'branch's Subvenance (in the case of an update), and
		// 3) the Subvenance of all branches in the *old* provenance of 'branch's Subvenance
		toUpdate := []*pfs.BranchInfo{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
			}
			toUpdate = append(toUpdate, subvBranchInfo)
		}
		// Sorting is important here because it sorts topologically. This means
		// that when evaluating element i of `toUpdate` all elements < i will
		// have already been evaluated and thus we can safely use their
		// Provenance field.
		sort.Slice(toUpdate, func(i, j int) bool { return len(toUpdate[i].Provenance) < len(toUpdate[j].Provenance) })
		for _, branchInfo := range toUpdate {
			oldProvenance := branchInfo.Provenance
			branchInfo.Provenance = nil
			// Re-compute Provenance
			for _, provBranch := range branchInfo.DirectProvenance {
				if err := d.addBranchProvenance(branchInfo, provBranch, stm); err != nil {
					return err
				}
				provBranchInfo := &pfs.BranchInfo{}
				if err := d.branches(provBranch.Repo.Name).ReadWrite(stm).Get(provBranch.Name, provBranchInfo); err != nil {
					return err
				}
				for _, provBranch := range provBranchInfo.Provenance {
					if err := d.addBranchProvenance(branchInfo, provBranch, stm); err != nil {
						return err
					}
				}
			}
			if err := d.branches(branchInfo.Branch.Repo.Name).ReadWrite(stm).Put(branchInfo.Branch.Name, branchInfo); err != nil {
				return err
			}
			// Update Subvenance of 'branchInfo's Provenance (incl. all Subvenance)
			for _, oldProvBranch := range oldProvenance {
				if !has(&branchInfo.Provenance, oldProvBranch) {
					// Provenance was deleted, so we delete ourselves from their subvenance
					oldProvBranchInfo := &pfs.BranchInfo{}
					if err := d.branches(oldProvBranch.Repo.Name).ReadWrite(stm).Update(oldProvBranch.Name, oldProvBranchInfo, func() error {
						del(&oldProvBranchInfo.Subvenance, branchInfo.Branch)
						return nil
					}); err != nil {
						return err
					}
				}
			}
		}

		// propagate the head commit to 'branch'. This may also modify 'branch', by
		// creating a new HEAD commit if 'branch's provenance was changed and its
		// current HEAD commit has old provenance
		return d.propagateCommit(stm, branch)
	})
	return err
}
func (d *driver) createBranch(pachClient *client.APIClient, branch *pfs.Branch, commit *pfs.Commit, provenance []*pfs.Branch) error {
	ctx := pachClient.Ctx()
	if err := d.checkIsAuthorized(pachClient, branch.Repo, auth.Scope_WRITER); err != nil {
		return err
	}
	
	
	
	if commit != nil {
		
		sameTarget := branch.Repo.Name == commit.Repo.Name && branch.Name == commit.ID
		if !sameTarget && provenance != nil {
			return fmt.Errorf("cannot point branch \"%s\" at target commit \"%s/%s\" without clearing its provenance",
				branch.Name, commit.Repo.Name, commit.ID)
		}
	}

	_, err := col.NewSTM(ctx, d.etcdClient, func(stm col.STM) error {
		
		var err error
		if commit != nil {
			_, err = d.resolveCommit(stm, commit) 
			if err != nil {
				
				
				if !isNoHeadErr(err) {
					return fmt.Errorf("unable to inspect %s/%s: %v", err, commit.Repo.Name, commit.ID)
				}
				commit = nil
			}
		}

		
		branches := d.branches(branch.Repo.Name).ReadWrite(stm)
		branchInfo := &pfs.BranchInfo{}
		if err := branches.Upsert(branch.Name, branchInfo, func() error {
			branchInfo.Name = branch.Name 
			branchInfo.Branch = branch
			branchInfo.Head = commit
			branchInfo.DirectProvenance = nil
			for _, provBranch := range provenance {
				add(&branchInfo.DirectProvenance, provBranch)
			}
			return nil
		}); err != nil {
			return err
		}
		repos := d.repos.ReadWrite(stm)
		repoInfo := &pfs.RepoInfo{}
		if err := repos.Update(branch.Repo.Name, repoInfo, func() error {
			add(&repoInfo.Branches, branch)
			return nil
		}); err != nil {
			return err
		}

		
		
		
		
		toUpdate := []*pfs.BranchInfo{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
			}
			toUpdate = append(toUpdate, subvBranchInfo)
		}
		
		
		
		
		sort.Slice(toUpdate, func(i, j int) bool { return len(toUpdate[i].Provenance) < len(toUpdate[j].Provenance) })
		for _, branchInfo := range toUpdate {
			oldProvenance := branchInfo.Provenance
			branchInfo.Provenance = nil
			
			for _, provBranch := range branchInfo.DirectProvenance {
				if err := d.addBranchProvenance(branchInfo, provBranch, stm); err != nil {
					return err
				}
				provBranchInfo := &pfs.BranchInfo{}
				if err := d.branches(provBranch.Repo.Name).ReadWrite(stm).Get(provBranch.Name, provBranchInfo); err != nil {
					return err
				}
				for _, provBranch := range provBranchInfo.Provenance {
					if err := d.addBranchProvenance(branchInfo, provBranch, stm); err != nil {
						return err
					}
				}
			}
			if err := d.branches(branchInfo.Branch.Repo.Name).ReadWrite(stm).Put(branchInfo.Branch.Name, branchInfo); err != nil {
				return err
			}
			
			for _, oldProvBranch := range oldProvenance {
				if !has(&branchInfo.Provenance, oldProvBranch) {
					
					oldProvBranchInfo := &pfs.BranchInfo{}
					if err := d.branches(oldProvBranch.Repo.Name).ReadWrite(stm).Update(oldProvBranch.Name, oldProvBranchInfo, func() error {
						del(&oldProvBranchInfo.Subvenance, branchInfo.Branch)
						return nil
					}); err != nil {
						return err
					}
				}
			}
		}

		
		
		
		return d.propagateCommit(stm, branch)
	})
	return err
}
func ( d * driver ) createBranch ( pachClient * client . APIClient , branch * pfs . Branch , commit * pfs . Commit , provenance [ ] * pfs . Branch ) error { ctx := pachClient . Ctx ( ) if err := d . checkIsAuthorized ( pachClient , branch . Repo , auth . Scope_WRITER ) ; err != nil { return err } // Validate request. The request must do exactly one of: // 1) updating 'branch's provenance (commit is nil OR commit == branch) // 2) re-pointing 'branch' at a new commit if commit != nil { // Determine if this is a provenance update sameTarget := branch . Repo . Name == commit . Repo . Name && branch . Name == commit . ID if ! sameTarget && provenance != nil { return fmt . Errorf ( " \" \" \" \" " , branch . Name , commit . Repo . Name , commit . ID ) } } _ , err := col . NewSTM ( ctx , d . etcdClient , func ( stm col . STM ) error { // if 'commit' is a branch, resolve it var err error if commit != nil { _ , err = d . resolveCommit ( stm , commit ) // if 'commit' is a branch, resolve it if err != nil { // possible that branch exists but has no head commit. This is fine, but // branchInfo.Head must also be nil if ! isNoHeadErr ( err ) { return fmt . Errorf ( " " , err , commit . Repo . Name , commit . ID ) } commit = nil } } // Retrieve (and create, if necessary) the current version of this branch branches := d . branches ( branch . Repo . Name ) . ReadWrite ( stm ) branchInfo := & pfs . BranchInfo { } if err := branches . Upsert ( branch . Name , branchInfo , func ( ) error { branchInfo . Name = branch . Name // set in case 'branch' is new branchInfo . Branch = branch branchInfo . Head = commit branchInfo . DirectProvenance = nil for _ , provBranch := range provenance { add ( & branchInfo . DirectProvenance , provBranch ) } return nil } ) ; err != nil { return err } repos := d . repos . ReadWrite ( stm ) repoInfo := & pfs . RepoInfo { } if err := repos . Update ( branch . Repo . Name , repoInfo , func ( ) error { add ( & repoInfo . Branches , branch ) return nil } ) ; err != nil { return err } // Update (or create) // 1) 'branch's Provenance // 2) the Provenance of all branches in 'branch's Subvenance (in the case of an update), and // 3) the Subvenance of all branches in the *old* provenance of 'branch's Subvenance toUpdate := [ ] * pfs . BranchInfo { 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 } toUpdate = append ( toUpdate , subvBranchInfo ) } // Sorting is important here because it sorts topologically. This means // that when evaluating element i of `toUpdate` all elements < i will // have already been evaluated and thus we can safely use their // Provenance field. sort . Slice ( toUpdate , func ( i , j int ) bool { return len ( toUpdate [ i ] . Provenance ) < len ( toUpdate [ j ] . Provenance ) } ) for _ , branchInfo := range toUpdate { oldProvenance := branchInfo . Provenance branchInfo . Provenance = nil // Re-compute Provenance for _ , provBranch := range branchInfo . DirectProvenance { if err := d . addBranchProvenance ( branchInfo , provBranch , stm ) ; err != nil { return err } provBranchInfo := & pfs . BranchInfo { } if err := d . branches ( provBranch . Repo . Name ) . ReadWrite ( stm ) . Get ( provBranch . Name , provBranchInfo ) ; err != nil { return err } for _ , provBranch := range provBranchInfo . Provenance { if err := d . addBranchProvenance ( branchInfo , provBranch , stm ) ; err != nil { return err } } } if err := d . branches ( branchInfo . Branch . Repo . Name ) . ReadWrite ( stm ) . Put ( branchInfo . Branch . Name , branchInfo ) ; err != nil { return err } // Update Subvenance of 'branchInfo's Provenance (incl. all Subvenance) for _ , oldProvBranch := range oldProvenance { if ! has ( & branchInfo . Provenance , oldProvBranch ) { // Provenance was deleted, so we delete ourselves from their subvenance oldProvBranchInfo := & pfs . BranchInfo { } if err := d . branches ( oldProvBranch . Repo . Name ) . ReadWrite ( stm ) . Update ( oldProvBranch . Name , oldProvBranchInfo , func ( ) error { del ( & oldProvBranchInfo . Subvenance , branchInfo . Branch ) return nil } ) ; err != nil { return err } } } } // propagate the head commit to 'branch'. This may also modify 'branch', by // creating a new HEAD commit if 'branch's provenance was changed and its // current HEAD commit has old provenance return d . propagateCommit ( stm , branch ) } ) return err }
// createBranch creates a new branch or updates an existing branch (must be one
// or the other). Most importantly, it sets 'branch.DirectProvenance' to
// 'provenance' and then for all (downstream) branches, restores the invariant:
//   ∀ b . b.Provenance = ∪ b'.Provenance (where b' ∈ b.DirectProvenance)
//
// This invariant is assumed to hold for all branches upstream of 'branch', but not
// for 'branch' itself once 'b.Provenance' has been set.
createBranch creates a new branch or updates an existing branch ( must be one or the other ) . Most importantly it sets branch . DirectProvenance to provenance and then for all ( downstream ) branches restores the invariant : ∀ b . b . Provenance = ∪ b . Provenance ( where b ∈ b . DirectProvenance ) This invariant is assumed to hold for all branches upstream of branch but not for branch itself once b . Provenance has been set .