diff --git a/synchronization/pipeline.h b/synchronization/pipeline.h
index 25ab0963df650bbfdc66dcc2cd11c41430e224e8..784b71ed977113f75508e736e02a3260d2b3b639 100644
--- a/synchronization/pipeline.h
+++ b/synchronization/pipeline.h
@@ -52,6 +52,8 @@ class Pipeline : public ftl::RefCountedThreadSafe<Pipeline<R>> {
       if (continuation_) {
         continuation_(nullptr, trace_id_);
         TRACE_EVENT_ASYNC_END0("flutter", "PipelineProduce", trace_id_);
+        // The continuation is being dropped on the floor. End the flow.
+        TRACE_FLOW_END("flutter", "PipelineItem", trace_id_);
       }
     }
 
@@ -60,6 +62,7 @@ class Pipeline : public ftl::RefCountedThreadSafe<Pipeline<R>> {
         continuation_(std::move(resource), trace_id_);
         continuation_ = nullptr;
         TRACE_EVENT_ASYNC_END0("flutter", "PipelineProduce", trace_id_);
+        TRACE_FLOW_STEP("flutter", "PipelineItem", trace_id_);
       }
     }
 
@@ -74,6 +77,7 @@ class Pipeline : public ftl::RefCountedThreadSafe<Pipeline<R>> {
 
     ProducerContinuation(Continuation continuation, size_t trace_id)
         : continuation_(continuation), trace_id_(trace_id) {
+      TRACE_FLOW_BEGIN("flutter", "PipelineItem", trace_id_);
       TRACE_EVENT_ASYNC_BEGIN0("flutter", "PipelineProduce", trace_id_);
     }
 
@@ -128,6 +132,8 @@ class Pipeline : public ftl::RefCountedThreadSafe<Pipeline<R>> {
 
     empty_.Signal();
 
+    TRACE_FLOW_END("flutter", "PipelineItem", trace_id);
+
     return items_count > 0 ? PipelineConsumeResult::MoreAvailable
                            : PipelineConsumeResult::Done;
   }