diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index 83a32591f..0e7fcae7f 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -526,6 +526,13 @@ static void add_pipeline_exact_delta_stats(yyjson_mut_doc *doc, yyjson_mut_val * if (stats.affected_paths >= 0) { yyjson_mut_obj_add_int(doc, exact, "affected_paths", stats.affected_paths); } + if (stats.affected_paths_limit >= 0) { + yyjson_mut_obj_add_int(doc, exact, "affected_paths_limit", + stats.affected_paths_limit); + } + if (stats.affected_paths_truncated) { + yyjson_mut_obj_add_bool(doc, exact, "affected_paths_truncated", true); + } if (stats.published_paths >= 0) { yyjson_mut_obj_add_int(doc, exact, "published_paths", stats.published_paths); } diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index e4e2c98ce..fc18fc84c 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -217,6 +217,8 @@ cbm_pipeline_t *cbm_pipeline_new(const char *repo_path, const char *db_path, p->exact_delta_stats.changed_paths = -1; p->exact_delta_stats.affected_paths = -1; p->exact_delta_stats.published_paths = -1; + p->exact_delta_stats.affected_paths_limit = -1; + p->exact_delta_stats.affected_paths_truncated = false; atomic_init(&p->cancelled, 0); return p; @@ -639,7 +641,7 @@ bool cbm_pipeline_incremental_derived_refresh_stale_on_exact(const cbm_pipeline_ } cbm_pipeline_exact_delta_stats_t cbm_pipeline_exact_delta_stats(const cbm_pipeline_t *p) { - static const cbm_pipeline_exact_delta_stats_t empty_stats = {-1, -1, -1}; + static const cbm_pipeline_exact_delta_stats_t empty_stats = {-1, -1, -1, -1, false}; return p ? p->exact_delta_stats : empty_stats; } @@ -685,12 +687,22 @@ void cbm_pipeline_set_publish_reason(cbm_pipeline_t *p, const char *reason) { void cbm_pipeline_set_exact_delta_stats(cbm_pipeline_t *p, int changed_paths, int affected_paths, int published_paths) { + cbm_pipeline_set_exact_delta_stats_with_limit(p, changed_paths, affected_paths, + published_paths, -1, false); +} + +void cbm_pipeline_set_exact_delta_stats_with_limit(cbm_pipeline_t *p, int changed_paths, + int affected_paths, int published_paths, + int affected_paths_limit, + bool affected_paths_truncated) { if (!p) { return; } p->exact_delta_stats.changed_paths = changed_paths; p->exact_delta_stats.affected_paths = affected_paths; p->exact_delta_stats.published_paths = published_paths; + p->exact_delta_stats.affected_paths_limit = affected_paths_limit; + p->exact_delta_stats.affected_paths_truncated = affected_paths_truncated; } static bool resolve_db_path_buf(const cbm_pipeline_t *p, char *path, size_t path_sz) { diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index b868ee646..68db77020 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -53,9 +53,11 @@ typedef enum { } cbm_pipeline_publish_kind_t; typedef struct { - int changed_paths; /* source paths classified as changed/deleted before frontier expansion */ - int affected_paths; /* paths in the exact-delta frontier, including changed/deleted paths */ - int published_paths; /* paths published by exact delta; 0 for exact no-op, -1 if not exact */ + int changed_paths; /* changed/deleted paths before frontier expansion */ + int affected_paths; /* exact frontier paths known before publish/fallback */ + int published_paths; /* paths published by exact delta; 0 for exact no-op, -1 if not exact */ + int affected_paths_limit; /* configured exact frontier cap when relevant; -1 if not reported */ + bool affected_paths_truncated; /* true when affected_paths reached the cap before full counting */ } cbm_pipeline_exact_delta_stats_t; /* Generation used by compatibility full/containment publishes that replace the diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 245b5d584..ca3f2fcba 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -1824,6 +1824,26 @@ cleanup: return CBM_STORE_OK; } +static void incr_report_exact_delta_plan_fallback(cbm_pipeline_t *p, + const cbm_pipeline_file_delta_plan_t *plan, + int input_path_count, int max_affected_paths, + const char *phase, + const char *default_reason) { + const char *reason = plan && plan->reason ? plan->reason : default_reason; + int affected_paths = (plan && plan->affected_count >= 0) ? plan->affected_count : -1; + cbm_pipeline_set_exact_delta_stats_with_limit(p, input_path_count, affected_paths, -1, + max_affected_paths, false); + cbm_pipeline_set_publish_reason(p, reason ? reason : "plan_error"); + if (affected_paths >= 0) { + cbm_log_info("incremental.exact.fallback", "reason", reason ? reason : "plan_error", + "phase", phase ? phase : "plan", "affected", itoa_buf_incr(affected_paths), + "max_affected", itoa_buf_incr(max_affected_paths)); + } else { + cbm_log_info("incremental.exact.fallback", "reason", reason ? reason : "plan_error", + "phase", phase ? phase : "plan"); + } +} + static int incr_try_exact_upsert_route(cbm_pipeline_t *p, cbm_store_t *store, const char *db_path, const char *project, cbm_file_info_t *changed_files, int changed_count, cbm_file_info_t *all_files, @@ -1879,11 +1899,25 @@ static int incr_try_exact_upsert_route(cbm_pipeline_t *p, cbm_store_t *store, co exact_files, &exact_count, exact_file_cap, &frontier_reason); if (frontier_rc != CBM_STORE_OK) { - cbm_pipeline_set_exact_delta_stats(p, input_path_count, exact_count + deleted_count, -1); + bool frontier_truncated = + frontier_reason && strcmp(frontier_reason, + CBM_PIPELINE_DELTA_REASON_FRONTIER_TOO_LARGE) == 0; + cbm_pipeline_set_exact_delta_stats_with_limit( + p, input_path_count, exact_count + deleted_count, -1, max_affected_paths, + frontier_truncated); cbm_pipeline_set_publish_reason( p, frontier_reason ? frontier_reason : CBM_PIPELINE_DELTA_REASON_FRONTIER_ERROR); - cbm_log_info("incremental.exact.fallback", "reason", - frontier_reason ? frontier_reason : CBM_PIPELINE_DELTA_REASON_FRONTIER_ERROR); + if (frontier_truncated) { + cbm_log_info("incremental.exact.fallback", "reason", + frontier_reason ? frontier_reason + : CBM_PIPELINE_DELTA_REASON_FRONTIER_ERROR, + "affected", itoa_buf_incr(exact_count + deleted_count), + "max_affected", itoa_buf_incr(max_affected_paths), "truncated", "true"); + } else { + cbm_log_info("incremental.exact.fallback", "reason", + frontier_reason ? frontier_reason + : CBM_PIPELINE_DELTA_REASON_FRONTIER_ERROR); + } free(exact_files); return CBM_STORE_OK; } @@ -2076,9 +2110,8 @@ static int incr_try_exact_upsert_route(cbm_pipeline_t *p, cbm_store_t *store, co max_affected_paths, &plan); CBM_PROF_END_N("incremental_exact", "10_plan_delta", t_exact_plan, delta_count); if (rc != CBM_STORE_OK || plan.route != CBM_PIPELINE_DELTA_ROUTE_EXACT_CANDIDATE) { - cbm_pipeline_set_publish_reason(p, plan.reason ? plan.reason : "plan_error"); - cbm_log_info("incremental.exact.fallback", "reason", - plan.reason ? plan.reason : "plan_error"); + incr_report_exact_delta_plan_fallback(p, &plan, input_path_count, max_affected_paths, + "plan", "plan_error"); goto cleanup; } cbm_pipeline_file_delta_plan_free(&plan); @@ -2172,9 +2205,8 @@ static int incr_try_exact_upsert_route(cbm_pipeline_t *p, cbm_store_t *store, co CBM_PROF_END_N("incremental_exact", "13_apply_delta", t_exact_apply, delta_count); if (rc != CBM_STORE_OK || plan.route != CBM_PIPELINE_DELTA_ROUTE_EXACT_CANDIDATE) { incr_mark_generation_failed(store, project, generation); - cbm_pipeline_set_publish_reason(p, plan.reason ? plan.reason : "apply_error"); - cbm_log_info("incremental.exact.fallback", "reason", - plan.reason ? plan.reason : "apply_error"); + incr_report_exact_delta_plan_fallback(p, &plan, input_path_count, max_affected_paths, + "apply", "apply_error"); goto cleanup; } diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index 3762d1e99..8a83d3189 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -965,6 +965,10 @@ bool cbm_pipeline_overlay_publish_small_deltas(const cbm_pipeline_t *p); bool cbm_pipeline_incremental_derived_refresh_stale_on_exact(const cbm_pipeline_t *p); void cbm_pipeline_set_exact_delta_stats(cbm_pipeline_t *p, int changed_paths, int affected_paths, int published_paths); +void cbm_pipeline_set_exact_delta_stats_with_limit(cbm_pipeline_t *p, int changed_paths, + int affected_paths, int published_paths, + int affected_paths_limit, + bool affected_paths_truncated); /* Parse a gRPC stub call "." into the canonical proto * service name + method. Returns true ONLY when a recognized gRPC stub/client diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index ff42340d0..91af4effc 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -10153,6 +10153,12 @@ TEST(incremental_fast_falls_back_for_oversized_inbound_frontier_and_matches_full ASSERT(strstr(logs, "msg=incremental.exact.fallback reason=frontier_too_large") != NULL); ASSERT_EQ(cbm_pipeline_publish_kind(p), CBM_PIPELINE_PUBLISH_INCREMENTAL_CONTAINMENT); ASSERT_STR_EQ(cbm_pipeline_publish_reason(p), "frontier_too_large"); + cbm_pipeline_exact_delta_stats_t stats = cbm_pipeline_exact_delta_stats(p); + ASSERT_EQ(stats.changed_paths, 1); + ASSERT_GT(stats.affected_paths, 1); + ASSERT_EQ(stats.affected_paths_limit, cbm_pipeline_exact_max_affected_paths(p)); + ASSERT_TRUE(stats.affected_paths_truncated); + ASSERT_EQ(stats.published_paths, -1); cbm_pipeline_free(p); cbm_store_t *owner_store = cbm_store_open_path(g_incr_dbpath);