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 the cleanup had
already deleted never got a stop event when it finished, and a billing
consumer that had recorded the start billed the task forever.

Now the stop is written when either piece of recorded start evidence exists:
the TASK_STARTED event, or the TASK_WAS_RUNNING baseline the backfill seeds
for tasks that were already running when the keep-running cleanup was
introduced. A legitimately started task always has one of the two: the
cleanup no longer deletes the start event of a running task, and the
backfill covers tasks that had already lost theirs. When neither exists (say
a task canceled before it ever ran), no consumer ever saw the task start,
and a stop event would be noise nothing can pair with.

The after_destroy hook now goes through the same check. It used to write a
stop unconditionally, so destroying a never-started PENDING task (app
deletion destroys each non-terminal task) produced exactly the unmatched
stop the update path avoids. Both pieces of evidence are looked up in one
query, and a comment pins a MySQL constraint: at MySQL's default REPEATABLE
READ isolation level, the evidence read must be the first query in the
surrounding transaction.

Author: joyvuu-dave

app/models/runtime/task_model.rb

@@ -42,7 +42,7 @@ def after_update
 
     def after_destroy
       super
-      create_stop_event unless terminal_state?
+      create_stop_event_if_needed unless terminal_state?
     end
 
     def run_action_user
@@ -137,9 +137,25 @@ 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: Repositories::AppUsageEventRepository::TASK_STOPPED_EVENT_STATE).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.
+      #
+      # NOTE: on MySQL (default REPEATABLE READ) these must be the first reads
+      # in the surrounding transaction. MySQL freezes what a transaction can
+      # see at its first read; if an earlier hook ran a query first, a baseline
+      # committed in the meantime would be invisible here, and the stop would
+      # be wrongly skipped.
+      start_evidence_states = [
+        Repositories::AppUsageEventRepository::TASK_STARTED_EVENT_STATE,
+        Repositories::AppUsageEventRepository::TASK_WAS_RUNNING_EVENT_STATE
+      ]
+      started = app_usage_repo.find_by_task_and_state(task: self, state: start_evidence_states)
+      return if started.nil?
 
       create_stop_event
     end

spec/unit/actions/task_delete_spec.rb

@@ -61,17 +61,23 @@ module VCAP::CloudController
         expect(events[0].metadata['task_guid']).to eq(task4.guid)
       end
 
-      it 'creates a usage event for non-terminal tasks' do
+      it 'creates a usage event for non-terminal tasks with recorded start evidence' do
+        # task4 and task5 wrote TASK_STARTED when they moved to RUNNING. task3
+        # is still PENDING: no consumer ever saw it start, so destroying it
+        # must not write a TASK_STOPPED that nothing can pair with a start.
+        create(:app_usage_event, task_guid: task4.guid, state: 'TASK_STARTED')
+        create(:app_usage_event, task_guid: task5.guid, state: 'TASK_STARTED')
+
         task_delete.delete_for_app(app.guid)
 
-        events = AppUsageEvent.where(parent_app_guid: app.guid).all
-        expect(events.size).to eq(3)
-        task_guids = [task3.guid, task4.guid, task5.guid]
+        events = AppUsageEvent.where(parent_app_guid: app.guid, state: 'TASK_STOPPED').all
+        expect(events.size).to eq(2)
+        task_guids = [task4.guid, task5.guid]
         events.each do |event|
-          expect(event.state).to eq('TASK_STOPPED')
           expect(task_guids.delete(event.task_guid)).not_to be_nil
         end
         expect(task_guids).to be_empty
+        expect(AppUsageEvent.where(task_guid: task3.guid, state: 'TASK_STOPPED')).to be_empty
       end
     end
   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, since no consumer ever saw the task 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, since no consumer ever saw the task 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
@@ -72,6 +118,17 @@ module VCAP::CloudController
             end
           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 the task does not have a TASK_STARTED event' do
             let!(:start_event) { nil }
 
@@ -132,7 +189,8 @@ module VCAP::CloudController
     end
 
     describe 'after destroy' do
-      let(:task) { create(:task_model, app: parent_app, state: TaskModel::PENDING_STATE) }
+      let(:task) { create(:task_model, app: parent_app, state: TaskModel::RUNNING_STATE) }
+      let!(:start_event) { create(:app_usage_event, task_guid: task.guid, state: 'TASK_STARTED') }
 
       it 'creates a TASK_STOPPED event' do
         task.destroy
@@ -143,6 +201,42 @@ module VCAP::CloudController
         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.destroy
+
+          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(:task) { create(:task_model, app: parent_app, state: TaskModel::PENDING_STATE) }
+        let!(:start_event) { nil }
+
+        it 'does not create a TASK_STOPPED event, since no consumer ever saw the task start' do
+          task.destroy
+
+          event = AppUsageEvent.find(task_guid: task.guid, state: 'TASK_STOPPED')
+          expect(event).to be_nil
+        end
+      end
+
+      context 'when the task already has a TASK_STOPPED event' do
+        before do
+          create(:app_usage_event, task_guid: task.guid, state: 'TASK_STOPPED')
+        end
+
+        it 'does not create an additional TASK_STOPPED event' do
+          task.destroy
+
+          events = AppUsageEvent.where(task_guid: task.guid, state: 'TASK_STOPPED')
+          expect(events.count).to equal(1)
+        end
+      end
+
       context 'when the task is already in a terminal state (and thus already has a stop event)' do
         describe 'when the task is failed' do
           let(:task) { create(:task_model, app: parent_app, state: TaskModel::FAILED_STATE) }