fix(ingestion): pass ShutdownSignal into issue and MR pagination loops
The orchestrator already accepted a ShutdownSignal but only checked it between phases (after all issues fetched, before discussions). The inner loops in ingest_issues() and ingest_merge_requests() consumed entire paginated streams without checking for cancellation. On a large initial sync (thousands of issues/MRs), Ctrl+C could be unresponsive for minutes while the current entity type finished draining. Now both functions accept &ShutdownSignal and check is_cancelled() at the top of each iteration, breaking out promptly and committing the cursor for whatever was already processed. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -7,6 +7,7 @@ use tracing::{debug, info, warn};
|
|||||||
use crate::Config;
|
use crate::Config;
|
||||||
use crate::core::error::{LoreError, Result};
|
use crate::core::error::{LoreError, Result};
|
||||||
use crate::core::payloads::{StorePayloadOptions, store_payload};
|
use crate::core::payloads::{StorePayloadOptions, store_payload};
|
||||||
|
use crate::core::shutdown::ShutdownSignal;
|
||||||
use crate::core::time::now_ms;
|
use crate::core::time::now_ms;
|
||||||
use crate::documents::SourceType;
|
use crate::documents::SourceType;
|
||||||
use crate::gitlab::GitLabClient;
|
use crate::gitlab::GitLabClient;
|
||||||
@@ -41,6 +42,7 @@ pub async fn ingest_issues(
|
|||||||
config: &Config,
|
config: &Config,
|
||||||
project_id: i64,
|
project_id: i64,
|
||||||
gitlab_project_id: i64,
|
gitlab_project_id: i64,
|
||||||
|
signal: &ShutdownSignal,
|
||||||
) -> Result<IngestIssuesResult> {
|
) -> Result<IngestIssuesResult> {
|
||||||
let mut result = IngestIssuesResult::default();
|
let mut result = IngestIssuesResult::default();
|
||||||
|
|
||||||
@@ -58,6 +60,10 @@ pub async fn ingest_issues(
|
|||||||
let mut last_gitlab_id: Option<i64> = None;
|
let mut last_gitlab_id: Option<i64> = None;
|
||||||
|
|
||||||
while let Some(issue_result) = issues_stream.next().await {
|
while let Some(issue_result) = issues_stream.next().await {
|
||||||
|
if signal.is_cancelled() {
|
||||||
|
info!("Issue ingestion interrupted by shutdown signal");
|
||||||
|
break;
|
||||||
|
}
|
||||||
let issue = issue_result?;
|
let issue = issue_result?;
|
||||||
result.fetched += 1;
|
result.fetched += 1;
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ use tracing::{debug, info, warn};
|
|||||||
use crate::Config;
|
use crate::Config;
|
||||||
use crate::core::error::{LoreError, Result};
|
use crate::core::error::{LoreError, Result};
|
||||||
use crate::core::payloads::{StorePayloadOptions, store_payload};
|
use crate::core::payloads::{StorePayloadOptions, store_payload};
|
||||||
|
use crate::core::shutdown::ShutdownSignal;
|
||||||
use crate::core::time::now_ms;
|
use crate::core::time::now_ms;
|
||||||
use crate::documents::SourceType;
|
use crate::documents::SourceType;
|
||||||
use crate::gitlab::GitLabClient;
|
use crate::gitlab::GitLabClient;
|
||||||
@@ -42,6 +43,7 @@ pub async fn ingest_merge_requests(
|
|||||||
project_id: i64,
|
project_id: i64,
|
||||||
gitlab_project_id: i64,
|
gitlab_project_id: i64,
|
||||||
full_sync: bool,
|
full_sync: bool,
|
||||||
|
signal: &ShutdownSignal,
|
||||||
) -> Result<IngestMergeRequestsResult> {
|
) -> Result<IngestMergeRequestsResult> {
|
||||||
let mut result = IngestMergeRequestsResult::default();
|
let mut result = IngestMergeRequestsResult::default();
|
||||||
|
|
||||||
@@ -58,6 +60,10 @@ pub async fn ingest_merge_requests(
|
|||||||
let per_page = 100u32;
|
let per_page = 100u32;
|
||||||
|
|
||||||
loop {
|
loop {
|
||||||
|
if signal.is_cancelled() {
|
||||||
|
info!("MR ingestion interrupted by shutdown signal");
|
||||||
|
break;
|
||||||
|
}
|
||||||
let page_result = client
|
let page_result = client
|
||||||
.fetch_merge_requests_page(
|
.fetch_merge_requests_page(
|
||||||
gitlab_project_id,
|
gitlab_project_id,
|
||||||
|
|||||||
@@ -119,7 +119,8 @@ pub async fn ingest_project_issues_with_progress(
|
|||||||
};
|
};
|
||||||
|
|
||||||
emit(ProgressEvent::IssuesFetchStarted);
|
emit(ProgressEvent::IssuesFetchStarted);
|
||||||
let issue_result = ingest_issues(conn, client, config, project_id, gitlab_project_id).await?;
|
let issue_result =
|
||||||
|
ingest_issues(conn, client, config, project_id, gitlab_project_id, signal).await?;
|
||||||
|
|
||||||
result.issues_fetched = issue_result.fetched;
|
result.issues_fetched = issue_result.fetched;
|
||||||
result.issues_upserted = issue_result.upserted;
|
result.issues_upserted = issue_result.upserted;
|
||||||
@@ -329,6 +330,7 @@ pub async fn ingest_project_merge_requests_with_progress(
|
|||||||
project_id,
|
project_id,
|
||||||
gitlab_project_id,
|
gitlab_project_id,
|
||||||
full_sync,
|
full_sync,
|
||||||
|
signal,
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user