Emit TASK_STOPPED usage events based on recorded start evidence

create_stop_event_if_needed skipped the TASK_STOPPED event whenever the
TASK_STARTED event was absent -- so a task whose start event had been pruned
never got a stop event on completion, and a billing consumer that recorded
the start would bill the task forever.

Emit the stop when either piece of recorded start evidence exists: the
TASK_STARTED event, or the TASK_WAS_RUNNING baseline seeded by the backfill
for tasks that were already running when the keep-running cleanup was
introduced. Between the two, a legitimately started task always has one --
the cleanup no longer prunes the start event of a running task, and the
backfill covers tasks that had already lost theirs. When neither exists
(e.g. a task canceled before it ever ran), no consumer ever saw the task
start, so a stop event would be unmatched noise.

Author: joyvuu-dave

app/models/runtime/task_model.rb

@@ -137,9 +137,16 @@ def create_start_event
     def create_stop_event_if_needed
       app_usage_repo = Repositories::AppUsageEventRepository.new
 
-      start_event = app_usage_repo.find_by_task_and_state(task: self, state: 'TASK_STARTED')
-      existing_stop_event = app_usage_repo.find_by_task_and_state(task: self, state: 'TASK_STOPPED')
-      return if start_event.nil? || existing_stop_event.present?
+      return if app_usage_repo.find_by_task_and_state(task: self, state: 'TASK_STOPPED').present?
+
+      # Record the stop only when there is recorded evidence that the task
+      # started: the TASK_STARTED event, or the TASK_WAS_RUNNING baseline seeded
+      # for tasks that were already running when the keep-running cleanup was
+      # introduced. Without either, no consumer ever saw the task start, so a
+      # stop event would be unmatched noise.
+      started = app_usage_repo.find_by_task_and_state(task: self, state: 'TASK_STARTED') ||
+                app_usage_repo.find_by_task_and_state(task: self, state: 'TASK_WAS_RUNNING')
+      return if started.nil?
 
       create_stop_event
     end

spec/unit/models/runtime/task_model_spec.rb

@@ -28,6 +28,30 @@ module VCAP::CloudController
           expect(event.task_guid).to eq(task.guid)
           expect(event.parent_app_guid).to eq(task.app.guid)
         end
+
+        context 'when the TASK_STARTED event has been pruned and a TASK_WAS_RUNNING baseline exists' do
+          let!(:start_event) { create(:app_usage_event, task_guid: task.guid, state: 'TASK_WAS_RUNNING') }
+
+          it 'still creates a TASK_STOPPED event' do
+            task.update(state: TaskModel::SUCCEEDED_STATE)
+
+            event = AppUsageEvent.find(task_guid: task.guid, state: 'TASK_STOPPED')
+            expect(event).not_to be_nil
+            expect(event.task_guid).to eq(task.guid)
+            expect(event.parent_app_guid).to eq(task.app.guid)
+          end
+        end
+
+        context 'when there is neither a TASK_STARTED event nor a TASK_WAS_RUNNING baseline' do
+          let!(:start_event) { nil }
+
+          it 'does not create a TASK_STOPPED event no consumer could match to a start' do
+            task.update(state: TaskModel::SUCCEEDED_STATE)
+
+            event = AppUsageEvent.find(task_guid: task.guid, state: 'TASK_STOPPED')
+            expect(event).to be_nil
+          end
+        end
       end
 
       context 'when the task is moving to the FAILED_STATE' do
@@ -43,6 +67,28 @@ module VCAP::CloudController
             expect(event.task_guid).to eq(task.guid)
             expect(event.parent_app_guid).to eq(task.app.guid)
           end
+
+          context 'when the TASK_STARTED event has been pruned and a TASK_WAS_RUNNING baseline exists' do
+            let!(:start_event) { create(:app_usage_event, task_guid: task.guid, state: 'TASK_WAS_RUNNING') }
+
+            it 'still creates a TASK_STOPPED event' do
+              task.update(state: TaskModel::FAILED_STATE)
+
+              event = AppUsageEvent.find(task_guid: task.guid, state: 'TASK_STOPPED')
+              expect(event).not_to be_nil
+            end
+          end
+
+          context 'when there is neither a TASK_STARTED event nor a TASK_WAS_RUNNING baseline' do
+            let!(:start_event) { nil }
+
+            it 'does not create a TASK_STOPPED event no consumer could match to a start' do
+              task.update(state: TaskModel::FAILED_STATE)
+
+              event = AppUsageEvent.find(task_guid: task.guid, state: 'TASK_STOPPED')
+              expect(event).to be_nil
+            end
+          end
         end
 
         context 'when the task is moving from the PENDING state' do