Update trigger comment in place instead of posting new comments

.github/workflows/ci.yml

@@ -41,6 +41,16 @@ jobs:
           node-version: '20'
           cache: 'npm'
 
+      - uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.GCP_SERVICE_ACCOUNT_EMAIL }}
+
+      - uses: google-github-actions/setup-gcloud@v2
+
+      - name: Install gke-gcloud-auth-plugin
+        run: gcloud components install gke-gcloud-auth-plugin
+
       - name: Install dependencies
         run: npm ci
 
@@ -50,6 +60,13 @@ jobs:
           requested-token-type: urn:pulumi:token-type:access_token:personal
           scope: user:${{ vars.PULUMI_USER }}
 
+      - name: Configure Pulumi stack
+        working-directory: infra
+        run: |
+          pulumi stack select --create ${{ vars.PULUMI_ORG }}/dev
+          pulumi config set gcp:project ${{ vars.GCP_PROJECT_ID }}
+          pulumi config set gcp:region ${{ vars.GCP_REGION }}
+
       - uses: pulumi/actions@v5
         with:
           command: preview
@@ -70,6 +87,15 @@ jobs:
           node-version: '20'
           cache: 'npm'
 
+      - uses: google-github-actions/auth@v2
+        with:
+          workload_identity_provider: ${{ vars.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ vars.GCP_SERVICE_ACCOUNT_EMAIL }}
+
+      - uses: google-github-actions/setup-gcloud@v2
+        with:
+          install_components: gke-gcloud-auth-plugin
+
       - name: Install dependencies
         run: npm ci
 
@@ -79,6 +105,19 @@ jobs:
           requested-token-type: urn:pulumi:token-type:access_token:personal
           scope: user:${{ vars.PULUMI_USER }}
 
+      - name: Configure Pulumi stack
+        working-directory: services
+        run: |
+          pulumi stack select --create ${{ vars.PULUMI_ORG }}/dev
+          pulumi config set gcp:project ${{ vars.GCP_PROJECT_ID }}
+          pulumi config set gcp:region ${{ vars.GCP_REGION }}
+          pulumi config set --secret githubToken '${{ secrets.GH_CONTROLLER_TOKEN }}'
+          if [ -n "${{ secrets.LOGFIRE_TOKEN }}" ]; then
+            pulumi config set --secret logfireToken '${{ secrets.LOGFIRE_TOKEN }}'
+          fi
+          pulumi config set imageTag ${{ github.sha }}
+          pulumi config set runnerRepoUrl '${{ github.server_url }}/${{ github.repository }}'
+
       - uses: pulumi/actions@v5
         with:
           command: preview

controller/migrations/004_section_body.sql

@@ -0,0 +1,5 @@
+-- Per-job markdown section body, written by the runner each time the job's
+-- visible state on the trigger comment changes (running → completed/failed).
+-- The controller renders the trigger comment by concatenating section_body
+-- across all jobs that share a comment_id.
+ALTER TABLE benchmark_jobs ADD COLUMN section_body TEXT;

controller/src/bin/controller.rs

@@ -34,12 +34,18 @@ async fn main() -> Result<()> {
 
     let token = CancellationToken::new();
     let ready = health::ready_flag();
+    let comment_locks = health::comment_locks();
+    let server_config = health::ServerConfig {
+        runner_repo_url: config.runner_repo_url.clone(),
+    };
 
     let health_handle = tokio::spawn(health::serve(
         token.clone(),
         ready.clone(),
         pool.clone(),
         gh.clone(),
+        comment_locks.clone(),
+        server_config,
     ));
 
     let poller = tokio::spawn(github_poller::poll_loop(
@@ -53,6 +59,7 @@ async fn main() -> Result<()> {
         config.clone(),
         pool.clone(),
         gh.clone(),
+        comment_locks.clone(),
         token.clone(),
     ));
 

controller/src/bin/runner.rs

@@ -6,7 +6,6 @@
 use anyhow::Result;
 use tracing::{error, info};
 
-use benchmark_controller::github;
 use benchmark_controller::runner::config::{BenchType, RunnerConfig};
 use benchmark_controller::runner::poster::CommentPoster;
 use benchmark_controller::runner::{bench_arrow, bench_criterion, bench_standard, shell};
@@ -67,27 +66,25 @@ async fn run_benchmark(config: &RunnerConfig, poster: &CommentPoster) -> Result<
 async fn post_error_comment(config: &RunnerConfig, poster: &CommentPoster) {
     let tail = shell::tail_log(20).await;
 
-    let footer = github::issues_footer(config.runner_repo_url.as_deref());
     let body = format!(
-        "Benchmark for [this request]({}) failed.\n\n\
+        "\u{1f6a8} **Benchmark failed**\n\n\
          Last 20 lines of output:\n\
          <details><summary>Click to expand</summary>\n\n\
          ```\n\
          {tail}\n\
          ```\n\n\
-         </details>{footer}",
-        config.comment_url,
+         </details>"
     );
 
-    let pr_number = match config.pr_number() {
+    let comment_id = match config.comment_id_i64() {
         Ok(n) => n,
         Err(e) => {
-            error!(error = %e, "cannot post error comment: failed to parse PR number");
+            error!(error = %e, "cannot post error section: failed to parse comment id");
             return;
         }
     };
 
-    if let Err(e) = poster.post_comment(&config.repo, pr_number, &body).await {
-        error!(error = %e, "failed to post error comment");
+    if let Err(e) = poster.update_section(&config.repo, comment_id, &body).await {
+        error!(error = %e, "failed to post error section");
     }
 }

controller/src/comment_render.rs

@@ -0,0 +1,127 @@
+//! Render the trigger comment body from per-job benchmark sections.
+//!
+//! Each runner job contributes one markdown "section" that describes its
+//! current state (running, completed, failed). Sibling jobs that share a
+//! trigger comment get their sections concatenated into a single block,
+//! delimited by HTML comment markers so we can re-extract the user's
+//! original body on subsequent updates.
+
+pub const MARKER_BEGIN: &str = "<!-- datafusion-benchmarking:begin -->";
+pub const MARKER_END: &str = "<!-- datafusion-benchmarking:end -->";
+
+/// Strip a previously-appended bot block from `body`, returning whatever the
+/// user originally wrote. If the markers aren't present (first edit, or
+/// human-edited comment), the entire body is treated as user content.
+pub fn extract_user_body(body: &str) -> &str {
+    match body.find(MARKER_BEGIN) {
+        Some(idx) => body[..idx].trim_end_matches(['\n', ' ']),
+        None => body.trim_end_matches(['\n', ' ']),
+    }
+}
+
+/// Build the full trigger-comment body: the user's original text, followed by
+/// the bot block containing one section per job. Returns `original_body`
+/// unchanged when there are no sections to render.
+pub fn render(original_body: &str, sections: &[String], footer: &str) -> String {
+    let user = extract_user_body(original_body);
+    if sections.iter().all(|s| s.trim().is_empty()) {
+        return user.to_string();
+    }
+    let mut out = String::with_capacity(user.len() + 256);
+    if !user.is_empty() {
+        out.push_str(user);
+        out.push_str("\n\n");
+    }
+    out.push_str(MARKER_BEGIN);
+    out.push('\n');
+    for (i, section) in sections.iter().enumerate() {
+        let trimmed = section.trim();
+        if trimmed.is_empty() {
+            continue;
+        }
+        if i > 0 {
+            out.push_str("\n\n");
+        }
+        out.push_str(trimmed);
+    }
+    if !footer.trim().is_empty() {
+        out.push_str("\n\n");
+        out.push_str(footer.trim_start_matches('\n'));
+    }
+    out.push('\n');
+    out.push_str(MARKER_END);
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn extract_returns_original_when_no_marker() {
+        assert_eq!(
+            extract_user_body("run benchmark tpch"),
+            "run benchmark tpch"
+        );
+    }
+
+    #[test]
+    fn extract_strips_bot_block() {
+        let body = format!("run benchmark tpch\n\n{MARKER_BEGIN}\nstuff\n{MARKER_END}",);
+        assert_eq!(extract_user_body(&body), "run benchmark tpch");
+    }
+
+    #[test]
+    fn extract_handles_marker_at_start() {
+        let body = format!("{MARKER_BEGIN}\nstuff\n{MARKER_END}");
+        assert_eq!(extract_user_body(&body), "");
+    }
+
+    #[test]
+    fn render_with_no_sections_returns_user_body() {
+        assert_eq!(render("hello", &[], ""), "hello");
+        assert_eq!(render("hello", &["".to_string()], ""), "hello");
+    }
+
+    #[test]
+    fn render_appends_single_section() {
+        let out = render("run benchmark tpch", &["section A".to_string()], "");
+        assert!(out.starts_with("run benchmark tpch\n\n"));
+        assert!(out.contains(MARKER_BEGIN));
+        assert!(out.contains("section A"));
+        assert!(out.trim_end().ends_with(MARKER_END));
+    }
+
+    #[test]
+    fn render_multiple_sections_separated() {
+        let out = render(
+            "trigger",
+            &["A".to_string(), "B".to_string(), "C".to_string()],
+            "",
+        );
+        let a = out.find('A').unwrap();
+        let b = out.find('B').unwrap();
+        let c = out.find('C').unwrap();
+        assert!(a < b && b < c);
+    }
+
+    #[test]
+    fn render_replaces_prior_bot_block() {
+        let prior = format!("trigger\n\n{MARKER_BEGIN}\nold section\n{MARKER_END}",);
+        let out = render(&prior, &["new section".to_string()], "");
+        assert!(out.contains("new section"));
+        assert!(!out.contains("old section"));
+        // Only one marker pair
+        assert_eq!(out.matches(MARKER_BEGIN).count(), 1);
+        assert_eq!(out.matches(MARKER_END).count(), 1);
+    }
+
+    #[test]
+    fn render_includes_footer_inside_block() {
+        let out = render("trigger", &["section".to_string()], "\n\n---\n[issue]");
+        let begin = out.find(MARKER_BEGIN).unwrap();
+        let end = out.find(MARKER_END).unwrap();
+        let issue = out.find("[issue]").unwrap();
+        assert!(begin < issue && issue < end);
+    }
+}

controller/src/db.rs

@@ -128,21 +128,52 @@ pub async fn set_runner_token(pool: &SqlitePool, job_id: i64, token: &str) -> Re
     Ok(())
 }
 
-/// Look up a running job by id and return its stored runner token, repo, and
-/// PR number. Returns `None` if the job doesn't exist.
+/// Look up a running job by id and return its stored runner token, repo, PR
+/// number, and the trigger comment id. Returns `None` if the job doesn't exist.
 pub async fn get_job_for_comment(
     pool: &SqlitePool,
     job_id: i64,
-) -> Result<Option<(String, i64, String, Option<String>)>> {
-    let row = sqlx::query_as::<_, (String, i64, String, Option<String>)>(
-        "SELECT repo, pr_number, status, runner_token FROM benchmark_jobs WHERE id = ?",
+) -> Result<Option<(String, i64, String, Option<String>, i64)>> {
+    let row = sqlx::query_as::<_, (String, i64, String, Option<String>, i64)>(
+        "SELECT repo, pr_number, status, runner_token, comment_id \
+         FROM benchmark_jobs WHERE id = ?",
     )
     .bind(job_id)
     .fetch_optional(pool)
     .await?;
     Ok(row)
 }
 
+/// Persist the markdown section a runner wants displayed on the trigger
+/// comment for this job. Replaces any prior section body for the same job.
+pub async fn set_section_body(pool: &SqlitePool, job_id: i64, body: &str) -> Result<()> {
+    sqlx::query(
+        "UPDATE benchmark_jobs SET section_body = ?, updated_at = datetime('now') WHERE id = ?",
+    )
+    .bind(body)
+    .bind(job_id)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Return all (job_id, section_body) pairs for jobs sharing a trigger comment,
+/// in insertion order, skipping jobs whose section hasn't been set yet.
+pub async fn get_sections_for_comment(
+    pool: &SqlitePool,
+    comment_id: i64,
+) -> Result<Vec<(i64, String)>> {
+    let rows = sqlx::query_as::<_, (i64, String)>(
+        "SELECT id, section_body FROM benchmark_jobs \
+         WHERE comment_id = ? AND section_body IS NOT NULL \
+         ORDER BY id",
+    )
+    .bind(comment_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows)
+}
+
 /// Return all jobs with status `running`.
 pub async fn get_active_jobs(pool: &SqlitePool) -> Result<Vec<BenchmarkJob>> {
     let jobs = sqlx::query_as::<_, BenchmarkJob>(
@@ -412,16 +443,63 @@ mod tests {
 
         set_runner_token(&pool, id, "secret-abc").await.unwrap();
 
-        let (repo, pr, status, token) = get_job_for_comment(&pool, id).await.unwrap().unwrap();
+        let (repo, pr, status, token, comment_id) =
+            get_job_for_comment(&pool, id).await.unwrap().unwrap();
         assert_eq!(repo, "apache/datafusion");
         assert_eq!(pr, 42);
         assert_eq!(status, "pending");
         assert_eq!(token.as_deref(), Some("secret-abc"));
+        assert_eq!(comment_id, 2000);
 
         // Missing job
         assert!(get_job_for_comment(&pool, 99_999).await.unwrap().is_none());
     }
 
+    // ── section_body helpers ────────────────────────────────────────
+
+    #[tokio::test]
+    async fn section_body_round_trip_and_aggregate() {
+        let pool = test_pool().await;
+        mark_comment_seen(&pool, 3000, "apache/datafusion", 42, "alice", "2024-01-01")
+            .await
+            .unwrap();
+        let id_a = insert_job(&pool, &test_job(3000)).await.unwrap();
+        let id_b = insert_job(&pool, &test_job(3000)).await.unwrap();
+
+        // Initially neither job has a section body
+        assert!(get_sections_for_comment(&pool, 3000)
+            .await
+            .unwrap()
+            .is_empty());
+
+        set_section_body(&pool, id_a, "running tpch").await.unwrap();
+        let sections = get_sections_for_comment(&pool, 3000).await.unwrap();
+        assert_eq!(sections, vec![(id_a, "running tpch".to_string())]);
+
+        set_section_body(&pool, id_b, "running clickbench")
+            .await
+            .unwrap();
+        // Replace job A's section body
+        set_section_body(&pool, id_a, "completed tpch")
+            .await
+            .unwrap();
+
+        let sections = get_sections_for_comment(&pool, 3000).await.unwrap();
+        assert_eq!(
+            sections,
+            vec![
+                (id_a, "completed tpch".to_string()),
+                (id_b, "running clickbench".to_string()),
+            ]
+        );
+
+        // A different comment_id should not see these sections
+        assert!(get_sections_for_comment(&pool, 9999)
+            .await
+            .unwrap()
+            .is_empty());
+    }
+
     // ── update_job_status + get_active_jobs ─────────────────────────
 
     #[tokio::test]