fix: queue stat should look for alldone marker and exit

Signed-off-by: Nick Mitchell <[email protected]>

pkg/observe/queuestreamer/stream.go

@@ -57,6 +57,9 @@ func once(ctx context.Context, c client, modelChan chan Model, doneChan chan str
 		fmt.Fprintf(os.Stderr, "Listen bucket=%s path=%s\n", c.RunContext.Bucket, prefix)
 	}
 
+	// We'll keep an eye out for this marker. If we see it, the run is done.
+	allDoneMarker := c.RunContext.AsFile(queue.AllDoneMarker)
+
 	objects, errs := c.s3.Listen(c.RunContext.Bucket, prefix, "", true)
 	defer c.s3.StopListening(c.RunContext.Bucket)
 
@@ -86,6 +89,13 @@ func once(ctx context.Context, c client, modelChan chan Model, doneChan chan str
 			if c.LogOptions.Verbose {
 				fmt.Fprintf(os.Stderr, "Queue streamer got push notification object=%s\n", obj)
 			}
+
+			if obj == allDoneMarker {
+				if c.LogOptions.Verbose {
+					fmt.Fprintln(os.Stderr, "Queue streamer got all done")
+				}
+				return nil
+			}
 		}
 
 		// fetch and parse model