Fix consistency VSA check worker

What does this MR do and why?

This MR fixes the VSA consistency check worker. Some items were not cleaned up due to a wrong filter scope on the query.

It turns out there is a bug in the ConsistencyCheckService as events don't get deleted when their associated record gets removed if the end event did not occur.

Note: users are not affected because we don't show items without end-events on the UI.

DB

Due to the filter changes, a new async index was added recently: !121669 (merged)

New query (50-800ms depending on the cache): https://console.postgres.ai/gitlab/gitlab-production-tunnel-pg12/sessions/19171/commands/63307 The query uses the in-operator optimization technique, the library generates a pretty large SQL query because one of the column no longer filters out null values and the column is also used in the ORDER BY clause. This means that the query needs to handle these cases:

• the end_event_timestamp is null
• the end_event_timestamp is not null

On PRD, all DB indexes have been created over the weekend.

Up:

main: == 20230605085936 AddNewIndexToVsaIssueStageEvents: migrating =================
main: -- transaction_open?()
main:    -> 0.0000s
main: -- index_name_exists?(:analytics_cycle_analytics_issue_stage_events, "index_issue_stage_events_for_consistency_check")
main:    -> 0.0052s
main: -- transaction_open?()
main:    -> 0.0000s
main: -- index_exists?("gitlab_partitions_static.analytics_cycle_analytics_issue_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :issue_id], {:name=>"index_c7ac8595d3", :algorithm=>:concurrently})
main:    -> 0.0045s
main: -- execute("SET statement_timeout TO 0")
main:    -> 0.0004s
main: -- add_index("gitlab_partitions_static.analytics_cycle_analytics_issue_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :issue_id], {:name=>"index_c7ac8595d3", :algorithm=>:concurrently})
... Same statements for the other partitions

main: == 20230605085957 AddNewIndexToVsaMrStageEvents: migrating ====================
main: -- transaction_open?()
main:    -> 0.0000s
main: -- index_name_exists?(:analytics_cycle_analytics_merge_request_stage_events, "index_mr_stage_events_for_consistency_check")
main:    -> 0.0019s
main: -- transaction_open?()
main:    -> 0.0000s
main: -- index_exists?("gitlab_partitions_static.analytics_cycle_analytics_merge_request_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :merge_request_id], {:name=>"index_ce87cbaf2d", :algorithm=>:concurrently})
main:    -> 0.0058s
main: -- add_index("gitlab_partitions_static.analytics_cycle_analytics_merge_request_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :merge_request_id], {:name=>"index_ce87cbaf2d", :algorithm=>:concurrently})
... Same statements for the other partitions



ci: == 20230605085936 AddNewIndexToVsaIssueStageEvents: migrating =================
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- index_name_exists?(:analytics_cycle_analytics_issue_stage_events, "index_issue_stage_events_for_consistency_check")
ci:    -> 0.0065s
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- index_exists?("gitlab_partitions_static.analytics_cycle_analytics_issue_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :issue_id], {:name=>"index_c7ac8595d3", :algorithm=>:concurrently})
ci:    -> 0.0072s
ci: -- execute("SET statement_timeout TO 0")
ci:    -> 0.0006s
ci: -- add_index("gitlab_partitions_static.analytics_cycle_analytics_issue_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :issue_id], {:name=>"index_c7ac8595d3", :algorithm=>:concurrently})
... Same statements for the other partitions

ci: == 20230605085957 AddNewIndexToVsaMrStageEvents: migrating ====================
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- index_name_exists?(:analytics_cycle_analytics_merge_request_stage_events, "index_mr_stage_events_for_consistency_check")
ci:    -> 0.0017s
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- index_exists?("gitlab_partitions_static.analytics_cycle_analytics_merge_request_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :merge_request_id], {:name=>"index_ce87cbaf2d", :algorithm=>:concurrently})
ci:    -> 0.0043s
ci: -- add_index("gitlab_partitions_static.analytics_cycle_analytics_merge_request_stage_events_00", [:stage_event_hash_id, :group_id, :end_event_timestamp, :merge_request_id], {:name=>"index_ce87cbaf2d", :algorithm=>:concurrently})
... Same statements for the other partitions

Down:

ci: == 20230605085957 AddNewIndexToVsaMrStageEvents: reverting ====================
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- index_name_exists?(:analytics_cycle_analytics_merge_request_stage_events, "index_mr_stage_events_for_consistency_check")
ci:    -> 0.0058s
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- remove_index(:analytics_cycle_analytics_merge_request_stage_events, {:name=>"index_mr_stage_events_for_consistency_check"})
ci:    -> 0.0042s
ci: == 20230605085957 AddNewIndexToVsaMrStageEvents: reverted (0.0866s) ===========

ci: == 20230605085936 AddNewIndexToVsaIssueStageEvents: reverting =================
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- index_name_exists?(:analytics_cycle_analytics_issue_stage_events, "index_issue_stage_events_for_consistency_check")
ci:    -> 0.0058s
ci: -- transaction_open?()
ci:    -> 0.0000s
ci: -- remove_index(:analytics_cycle_analytics_issue_stage_events, {:name=>"index_issue_stage_events_for_consistency_check"})
ci:    -> 0.0026s
ci: == 20230605085936 AddNewIndexToVsaIssueStageEvents: reverted (0.0851s) ========

main: == 20230605085957 AddNewIndexToVsaMrStageEvents: reverting ====================
main: -- transaction_open?()
main:    -> 0.0000s
main: -- index_name_exists?(:analytics_cycle_analytics_merge_request_stage_events, "index_mr_stage_events_for_consistency_check")
main:    -> 0.0059s
main: -- transaction_open?()
main:    -> 0.0000s
main: -- remove_index(:analytics_cycle_analytics_merge_request_stage_events, {:name=>"index_mr_stage_events_for_consistency_check"})
main:    -> 0.0043s
main: == 20230605085957 AddNewIndexToVsaMrStageEvents: reverted (0.0798s) ===========


main: == 20230605085936 AddNewIndexToVsaIssueStageEvents: reverting =================
main: -- transaction_open?()
main:    -> 0.0000s
main: -- index_name_exists?(:analytics_cycle_analytics_issue_stage_events, "index_issue_stage_events_for_consistency_check")
main:    -> 0.0053s
main: -- transaction_open?()
main:    -> 0.0000s
main: -- remove_index(:analytics_cycle_analytics_issue_stage_events, {:name=>"index_issue_stage_events_for_consistency_check"})
main:    -> 0.0026s
main: == 20230605085936 AddNewIndexToVsaIssueStageEvents: reverted (0.0749s) ========

MR acceptance checklist

This checklist encourages us to confirm any changes have been analyzed to reduce risks in quality, performance, reliability, security, and maintainability.

• I have evaluated the MR acceptance checklist for this MR.

Related to #408320 (closed)