src/server/pps/server/api_server.go
func (a *apiServer) hardStopPipeline(pachClient *client.APIClient, pipelineInfo *pps.PipelineInfo) error {
	// Remove the output branch's provenance so that no new jobs can be created
	if err := pachClient.CreateBranch(
		pipelineInfo.Pipeline.Name,
		pipelineInfo.OutputBranch,
		pipelineInfo.OutputBranch,
		nil,
	); err != nil && !isNotFoundErr(err) {
		return fmt.Errorf("could not recreate original output branch: %v", err)
	}

	// Now that new commits won't be created on the master branch, enumerate
	// existing commits and close any open ones.
	iter, err := pachClient.ListCommitStream(pachClient.Ctx(), &pfs.ListCommitRequest{
		Repo: client.NewRepo(pipelineInfo.Pipeline.Name),
		To:   client.NewCommit(pipelineInfo.Pipeline.Name, pipelineInfo.OutputBranch),
	})
	if err != nil {
		return fmt.Errorf("couldn't get open commits on '%s': %v", pipelineInfo.OutputBranch, err)
	}
	// Finish all open commits, most recent first (so that we finish the
	// current job's output commit--the oldest--last, and unblock the master
	// only after all other commits are also finished, preventing any new jobs)
	for {
		ci, err := iter.Recv()
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		if ci.Finished == nil {
			// Finish the commit and don't pass a tree
			pachClient.PfsAPIClient.FinishCommit(pachClient.Ctx(), &pfs.FinishCommitRequest{
				Commit: ci.Commit,
				Empty:  true,
			})
		}
	}
	return nil
}
func (a *apiServer) hardStopPipeline(pachClient *client.APIClient, pipelineInfo *pps.PipelineInfo) error {
	// Remove the output branch's provenance so that no new jobs can be created
	if err := pachClient.CreateBranch(
		pipelineInfo.Pipeline.Name,
		pipelineInfo.OutputBranch,
		pipelineInfo.OutputBranch,
		nil,
	); err != nil && !isNotFoundErr(err) {
		return fmt.Errorf("could not recreate original output branch: %v", err)
	}

	// Now that new commits won't be created on the master branch, enumerate
	// existing commits and close any open ones.
	iter, err := pachClient.ListCommitStream(pachClient.Ctx(), &pfs.ListCommitRequest{
		Repo: client.NewRepo(pipelineInfo.Pipeline.Name),
		To:   client.NewCommit(pipelineInfo.Pipeline.Name, pipelineInfo.OutputBranch),
	})
	if err != nil {
		return fmt.Errorf("couldn't get open commits on '%s': %v", pipelineInfo.OutputBranch, err)
	}
	// Finish all open commits, most recent first (so that we finish the
	// current job's output commit--the oldest--last, and unblock the master
	// only after all other commits are also finished, preventing any new jobs)
	for {
		ci, err := iter.Recv()
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		if ci.Finished == nil {
			// Finish the commit and don't pass a tree
			pachClient.PfsAPIClient.FinishCommit(pachClient.Ctx(), &pfs.FinishCommitRequest{
				Commit: ci.Commit,
				Empty:  true,
			})
		}
	}
	return nil
}
// hardStopPipeline does essentially the same thing as StopPipeline (deletes the
// pipeline's branch provenance, deletes any open commits, deletes any k8s
// workers), but does it immediately. This is to avoid races between operations
// that will do subsequent work (e.g. UpdatePipeline and DeletePipeline) and the
// PPS master