Backport Fix for Spout pipelines not able to restart

src/server/pps/cmds/cmds_test.go

@@ -203,7 +203,7 @@ func TestUnrunnableJobInfo(t *testing.T) {
 		"pipeline", pipeline2, "inputPipeline", pipeline1).Run())
 	require.NoError(t, tu.PachctlBashCmd(t, c, `
 		pachctl wait commit data@master
-		sleep 10
+		sleep 20
 		# make sure that there is a not-run job
 		pachctl list job --raw \
 			| match "JOB_UNRUNNABLE"

src/server/pps/server/api_server.go

@@ -1796,24 +1796,24 @@ func getExpectedNumWorkers(pipelineInfo *pps.PipelineInfo) (int, error) {
 // CreatePipeline implements the protobuf pps.CreatePipeline RPC
 //
 // Implementation note:
-// - CreatePipeline always creates pipeline output branches such that the
-//   pipeline's spec branch is in the pipeline output branch's provenance
-// - CreatePipeline will always create a new output commit, but that's done
-//   by CreateBranch at the bottom of the function, which sets the new output
-//   branch provenance, rather than commitPipelineInfoFromFileSet higher up.
-// - This is because CreatePipeline calls hardStopPipeline towards the top,
-// 	 breaking the provenance connection from the spec branch to the output branch
-// - For straightforward pipeline updates (e.g. new pipeline image)
-//   stopping + updating + starting the pipeline isn't necessary
-// - However it is necessary in many slightly atypical cases  (e.g. the
-//   pipeline input changed: if the spec commit is created while the
-//   output branch has its old provenance, or the output branch gets new
-//   provenance while the old spec commit is the HEAD of the spec branch,
-//   then an output commit will be created with provenance that doesn't
-//   match its spec's PipelineInfo.Details.Input. Another example is when
-//   request.Reprocess == true).
-// - Rather than try to enumerate every case where we can't create a spec
-//   commit without stopping the pipeline, we just always stop the pipeline
+//   - CreatePipeline always creates pipeline output branches such that the
+//     pipeline's spec branch is in the pipeline output branch's provenance
+//   - CreatePipeline will always create a new output commit, but that's done
+//     by CreateBranch at the bottom of the function, which sets the new output
+//     branch provenance, rather than commitPipelineInfoFromFileSet higher up.
+//   - This is because CreatePipeline calls hardStopPipeline towards the top,
+//     breaking the provenance connection from the spec branch to the output branch
+//   - For straightforward pipeline updates (e.g. new pipeline image)
+//     stopping + updating + starting the pipeline isn't necessary
+//   - However it is necessary in many slightly atypical cases  (e.g. the
+//     pipeline input changed: if the spec commit is created while the
+//     output branch has its old provenance, or the output branch gets new
+//     provenance while the old spec commit is the HEAD of the spec branch,
+//     then an output commit will be created with provenance that doesn't
+//     match its spec's PipelineInfo.Details.Input. Another example is when
+//     request.Reprocess == true).
+//   - Rather than try to enumerate every case where we can't create a spec
+//     commit without stopping the pipeline, we just always stop the pipeline
 func (a *apiServer) CreatePipeline(ctx context.Context, request *pps.CreatePipelineRequest) (response *types.Empty, retErr error) {
 	metricsFn := metrics.ReportUserAction(ctx, a.reporter, "CreatePipeline")
 	defer func(start time.Time) { metricsFn(start, retErr) }(time.Now())
@@ -2644,11 +2644,13 @@ func (a *apiServer) StartPipeline(ctx context.Context, request *pps.StartPipelin
 			return errors.EnsureStack(err)
 		}
 		// restore same provenance to meta repo
-		if err := a.env.PFSServer.CreateBranchInTransaction(txnCtx, &pfs.CreateBranchRequest{
-			Branch:     client.NewSystemRepo(pipelineInfo.Pipeline.Name, pfs.MetaRepoType).NewBranch(pipelineInfo.Details.OutputBranch),
-			Provenance: provenance,
-		}); err != nil {
-			return errors.EnsureStack(err)
+		if pipelineInfo.Details.Spout == nil && pipelineInfo.Details.Service == nil {
+			if err := a.env.PFSServer.CreateBranchInTransaction(txnCtx, &pfs.CreateBranchRequest{
+				Branch:     client.NewSystemRepo(pipelineInfo.Pipeline.Name, pfs.MetaRepoType).NewBranch(pipelineInfo.Details.OutputBranch),
+				Provenance: provenance,
+			}); err != nil {
+				return errors.EnsureStack(err)
+			}
 		}
 
 		newPipelineInfo := &pps.PipelineInfo{}

src/server/spout_test.go

@@ -557,4 +557,39 @@ func testSpout(t *testing.T, usePachctl bool) {
 		}))
 		require.NoError(t, c.DeleteAll())
 	})
+
+	t.Run("SpoutRestart", func(t *testing.T) {
+		pipeline := tu.UniqueString("pipeline")
+		_, err := c.PpsAPIClient.CreatePipeline(
+			c.Ctx(),
+			&pps.CreatePipelineRequest{
+				Pipeline: client.NewPipeline(pipeline),
+				Transform: &pps.Transform{
+					Cmd: []string{"sleep", "infinity"},
+				},
+				Spout: &pps.Spout{},
+			},
+		)
+		require.NoError(t, err)
+		require.NoError(t, backoff.Retry(func() error {
+			pi, err := c.InspectPipeline(pipeline, false)
+			require.NoError(t, err)
+			if pi.State != pps.PipelineState_PIPELINE_RUNNING {
+				return errors.Errorf("expected pipeline state: %s, but got: %s", pps.PipelineState_PIPELINE_RUNNING, pi.State)
+			}
+			return nil
+		}, backoff.NewTestingBackOff()))
+
+		// stop and start spout pipeline
+		require.NoError(t, c.StopPipeline(pipeline))
+		require.NoError(t, backoff.Retry(func() error {
+			pi, err := c.InspectPipeline(pipeline, false)
+			require.NoError(t, err)
+			if pi.State != pps.PipelineState_PIPELINE_PAUSED {
+				return errors.Errorf("expected pipeline state: %s, but got: %s", pps.PipelineState_PIPELINE_PAUSED, pi.State)
+			}
+			return nil
+		}, backoff.NewTestingBackOff()))
+		require.NoError(t, c.StartPipeline(pipeline))
+	})
 }