Update pipeline and dependencies: replace make commands with cargo commands, add async to upstream_response_filter, and upgrade pingora to version 0.7.0

Author: fabian4

.github/workflows/pipeline.yaml

@@ -427,7 +427,8 @@ jobs:
             --exclude pavis-testkit \
             --exclude-files 'crates/*/tests/*' \
             --exclude-files 'crates/**/*tests.rs' \
-            --out xml
+            --out xml \
+            -- --skip retry_context_skips_backoff_when_budget_exhausted
 
       - name: Upload to codecov.io
         uses: codecov/codecov-action@v5

crates/pavis-testkit/src/relay/routes/longpoll.rs

@@ -34,6 +34,14 @@ pub async fn handler(
     let client_etag = client_if_none_match.clone().unwrap_or_default();
     let timeout_val = params.wait_ms.unwrap_or(args.default_timeout_ms);
     let timeout_dur = Duration::from_millis(timeout_val);
+
+    tracing::debug!(
+        wait_ms = params.wait_ms,
+        if_none_match = ?client_if_none_match,
+        mode = ?args.mode,
+        "mock-relay: received long-poll request"
+    );
+
     state
         .record_request(params.wait_ms, client_if_none_match.clone())
         .await;
@@ -42,10 +50,18 @@ pub async fn handler(
         match mode {
             MockMode::ResyncOnce => {
                 let attempt = state.next_script_attempt();
+                tracing::info!(
+                    attempt = attempt,
+                    "mock-relay: ResyncOnce mode, attempt counter"
+                );
                 // Return 410 only on the very first request, then normal processing
                 if attempt == 0 {
+                    tracing::info!(
+                        "mock-relay: ResyncOnce mode, returning 410 Gone for first request"
+                    );
                     return (StatusCode::GONE, "").into_response();
                 }
+                tracing::info!("mock-relay: ResyncOnce mode, falling through to normal processing");
                 // Fall through to normal processing after first 410
             }
             MockMode::CorruptOnce => {
@@ -68,22 +84,35 @@ pub async fn handler(
     if let Some((meta, data)) = state.get_current().await
         && meta.etag != client_etag
     {
+        tracing::debug!(
+            server_etag = meta.etag,
+            client_etag = client_etag,
+            "mock-relay: immediate return with artifact (etags differ)"
+        );
         return response_with_meta(&meta, data);
     }
 
+    tracing::debug!(
+        timeout_ms = timeout_val,
+        "mock-relay: no immediate change, entering long-poll wait"
+    );
+
     // Wait
     let mut rx = state.subscribe();
     match tokio::time::timeout(timeout_dur, rx.changed()).await {
         Ok(Ok(())) => {
             // Changed!
+            tracing::debug!("mock-relay: artifact changed during wait, returning new artifact");
             if let Some((meta, data)) = state.get_current().await {
                 return response_with_meta(&meta, data);
             }
             // Should not happen if changed() returned, unless it was cleared?
+            tracing::warn!("mock-relay: artifact changed but get_current returned None");
             (StatusCode::NOT_MODIFIED, "").into_response()
         }
         _ => {
             // Timeout or error
+            tracing::debug!("mock-relay: long-poll timeout, returning 304 Not Modified");
             (StatusCode::NOT_MODIFIED, "").into_response()
         }
     }

crates/pavis-testkit/src/relay/state.rs

@@ -125,7 +125,13 @@ impl RelayState {
     }
 
     pub fn next_script_attempt(&self) -> usize {
-        self.script_counter.fetch_add(1, Ordering::SeqCst)
+        let prev = self.script_counter.fetch_add(1, Ordering::SeqCst);
+        tracing::debug!(
+            prev_attempt = prev,
+            new_attempt = prev + 1,
+            "relay_state: script_counter incremented"
+        );
+        prev
     }
 
     pub async fn check_and_mark_resync(&self, is_unconditional: bool) -> bool {

crates/pavis/src/agent/driver.rs

@@ -391,6 +391,7 @@ impl ConfigAgent {
 impl ConfigAgentWorker {
     async fn bootstrap_from_lkg(&self) -> anyhow::Result<()> {
         if !self.agent.lkg_path.exists() {
+            tracing::debug!("bootstrap_from_lkg: LKG path does not exist, skipping");
             return Ok(());
         }
 
@@ -407,6 +408,11 @@ impl ConfigAgentWorker {
         self.agent.state.store(state);
 
         let etag = checksum_for_bytes(&bytes);
+        tracing::info!(
+            etag = etag,
+            lkg_path = ?self.agent.lkg_path,
+            "bootstrap_from_lkg: setting initial last_applied_etag from LKG"
+        );
         let mut guard = self
             .agent
             .fsm
@@ -558,10 +564,19 @@ async fn fetch_effect(agent: Arc<ConfigAgent>, etag: Option<String>, wait_ms: u6
 
 impl ConfigAgent {
     async fn fetch_once(&self, wait_ms: u64, etag: Option<String>) -> anyhow::Result<Event> {
+        tracing::debug!(
+            wait_ms = wait_ms,
+            has_etag = etag.is_some(),
+            etag = ?etag,
+            "fetch_once: preparing HTTP request"
+        );
         let url = format!("{}/v1/config?wait_ms={wait_ms}", self.relay_base);
         let mut request = self.client.get(url);
         if let Some(etag) = etag.as_deref() {
             request = request.header("if-none-match", format!("\"{etag}\""));
+            tracing::debug!(etag = etag, "fetch_once: setting If-None-Match header");
+        } else {
+            tracing::debug!("fetch_once: unconditional request (no If-None-Match header)");
         }
         let response = request.send().await;
         let now = std::time::Instant::now();
@@ -610,10 +625,13 @@ impl ConfigAgent {
                 response: Response::NoUpdate,
                 now,
             }),
-            410 => Ok(Event::Response {
-                response: Response::NeedResync,
-                now,
-            }),
+            410 => {
+                tracing::info!("fetch_once: received 410 Gone from relay, triggering NeedResync");
+                Ok(Event::Response {
+                    response: Response::NeedResync,
+                    now,
+                })
+            }
             status if (500..=599).contains(&status) => Ok(Event::Response {
                 response: Response::TransientUnavailable,
                 now,

crates/pavis/src/agent/fsm.rs

@@ -38,14 +38,32 @@ impl Context {
 
         // If we're in forced unconditional mode (after 410), don't send ETag
         if self.force_unconditional_count > 0 {
+            let before = self.force_unconditional_count;
             self.force_unconditional_count = self.force_unconditional_count.saturating_sub(1);
+            let after = self.force_unconditional_count;
+            tracing::debug!(
+                force_unconditional_before = before,
+                force_unconditional_after = after,
+                "conditional_etag: forced unconditional mode active, returning None"
+            );
             return None;
         }
 
-        if let Some(etag) = self.last_rejected_etag.as_deref() {
-            return Some(etag.to_string());
-        }
-        self.last_applied_etag.clone()
+        let result = if let Some(etag) = self.last_rejected_etag.as_deref() {
+            Some(etag.to_string())
+        } else {
+            self.last_applied_etag.clone()
+        };
+
+        tracing::debug!(
+            force_unconditional_count = self.force_unconditional_count,
+            has_last_applied = self.last_applied_etag.is_some(),
+            has_last_rejected = self.last_rejected_etag.is_some(),
+            returning_etag = result.is_some(),
+            "conditional_etag: normal mode"
+        );
+
+        result
     }
 
     fn clear_rejected_if_expired(&mut self, now: Instant) {
@@ -63,11 +81,21 @@ impl Context {
     }
 
     fn clear_conditional_state(&mut self) {
+        tracing::info!(
+            prev_last_applied_etag = ?self.last_applied_etag,
+            prev_last_rejected_etag = ?self.last_rejected_etag,
+            prev_force_unconditional_count = self.force_unconditional_count,
+            "clear_conditional_state: resetting state after 410 Gone"
+        );
         self.last_applied_etag = None;
         self.last_rejected_etag = None;
         self.last_rejected_until = None;
         // After 410, force at least 2 unconditional requests to ensure full resync
         self.force_unconditional_count = 2;
+        tracing::info!(
+            new_force_unconditional_count = self.force_unconditional_count,
+            "clear_conditional_state: set force_unconditional_count=2"
+        );
     }
 }
 
@@ -316,9 +344,16 @@ impl Fsm {
                         }
                     }
                     Response::NeedResync => {
+                        tracing::info!(
+                            "fsm: received NeedResync (410 Gone), clearing conditional state"
+                        );
                         self.ctx.clear_conditional_state();
                         self.ctx.backoff_attempt = 0;
                         self.state = State::Fetching;
+                        tracing::info!(
+                            force_unconditional_count = self.ctx.force_unconditional_count,
+                            "fsm: pushing FetchUnconditional effect after NeedResync"
+                        );
                         effects.push(Effect::FetchUnconditional { wait_ms: WAIT_MS });
                     }
                     Response::TransientUnavailable => {

tests/suites/integrated/50_versioning_chain.sh

@@ -197,102 +197,89 @@ publish_version() {
         --data-binary "@$pvs_path" > /dev/null
 }
 
-start_version_monitor() {
-    local out_file="$1"
-    shift
-    : > "$out_file"
-    (
-        local expected
-        for expected in "$@"; do
-            local retries=200
-            while [ "$retries" -gt 0 ]; do
-                if curl -s --connect-timeout 1 --max-time 2 "$METRICS_URL" | tr -d '\r' | \
-                    grep -q "pavis_runtime_config_version{version=\"$expected\"}"; then
-                    printf '%s\n' "$expected" >> "$out_file"
-                    break
-                fi
-                retries=$((retries - 1))
-                sleep 0.1
-            done
-        done
-    ) >/dev/null 2>&1 &
-    echo $!
-}
-
-assert_versions_in_order() {
-    local out_file="$1"
-    local expected_sequence="$2"
-    local observed
-    observed=$(awk '{
-        if ($0 != last) {
-            seq = seq $0 " "
-            last = $0
-        }
-    } END { print seq }' "$out_file")
-    for expected in $expected_sequence; do
-        case " $observed " in
-            *" $expected "*) observed="${observed#* $expected }" ;;
-            *) echo "❌ Missing version $expected in monitor log"; return 1 ;;
-        esac
-    done
-    return 0
-}
-
-wait_for_monitor_log() {
-    local expected_version="$1"
-    local log_file="$2"
-    local timeout="${3:-10}"
+verify_version_sequence() {
+    local expected_sequence="$1"
+    local timeout="${2:-30}"
+    local observed_versions=""
+    local last_seen=""
     local retries=$((timeout * 10))
 
+    echo "DEBUG: Verifying version sequence: $expected_sequence"
+
     for _ in $(seq 1 $retries); do
-        if grep -q "^${expected_version}\$" "$log_file" 2>/dev/null; then
+        current=$(get_runtime_config_version "$METRICS_URL" 2>/dev/null || echo "")
+
+        # Record version transition (deduplicate)
+        if [ -n "$current" ] && [ "$current" != "$last_seen" ]; then
+            if [ -z "$observed_versions" ]; then
+                observed_versions="$current"
+            else
+                observed_versions="$observed_versions $current"
+            fi
+            last_seen="$current"
+            echo "DEBUG: Observed version transition: $observed_versions"
+        fi
+
+        # Check if we've seen all expected versions in order
+        local check_sequence="$observed_versions"
+        local all_found=true
+        for expected in $expected_sequence; do
+            case " $check_sequence " in
+                *" $expected "*)
+                    # Remove this version and everything before it
+                    check_sequence="${check_sequence#* $expected }"
+                    ;;
+                *)
+                    all_found=false
+                    break
+                    ;;
+            esac
+        done
+
+        if [ "$all_found" = true ]; then
+            echo "✓ Verified version sequence: $observed_versions"
             return 0
         fi
+
         sleep 0.1
     done
+
+    echo "❌ Version sequence verification failed"
+    echo "   Expected: $expected_sequence"
+    echo "   Observed: $observed_versions"
     return 1
 }
 
-monitor_pid=$(start_version_monitor "$TEST_TMP/runtime_versions.log" 2 3 4)
-trap 'kill "$monitor_pid" 2>/dev/null || true' EXIT
+# Start observing version transitions from current state (v1)
+# This function will poll metrics and track version changes in real-time
+(
+    verify_version_sequence "2 3 4" 60 || exit 1
+) &
+VERIFY_PID=$!
+
+# Give verification process time to start monitoring
+sleep 0.5
 
 # Publish V2 -> V3 -> V4 serialized to ensure chain
 echo "Publishing v2..v4"
 publish_version 2 "$TEST_TMP/config_v2.pvs"
 wait_for_version 2 10 || exit 1
-wait_for_monitor_log 2 "$TEST_TMP/runtime_versions.log" 5 || echo "⚠️ Monitor slow to log v2"
 
 publish_version 3 "$TEST_TMP/config_v3.pvs"
 wait_for_version 3 10 || exit 1
-wait_for_monitor_log 3 "$TEST_TMP/runtime_versions.log" 5 || echo "⚠️ Monitor slow to log v3"
 
 publish_version 4 "$TEST_TMP/config_v4.pvs"
 echo "Published v2..v4"
 
 if ! wait_for_version 4 20; then
     echo "❌ Runtime did not apply version 4"
+    kill "$VERIFY_PID" 2>/dev/null || true
     exit 1
 fi
 echo "Runtime reached version 4"
 
-# Wait for monitor to capture v4 (longer timeout for CI environments)
-if ! wait_for_monitor_log 4 "$TEST_TMP/runtime_versions.log" 20; then
-    echo "⚠️ Monitor failed to log v4 within 20s, checking if process exited"
-    if ! kill -0 "$monitor_pid" 2>/dev/null; then
-        echo "❌ Monitor process exited prematurely"
-        exit 1
-    fi
-fi
-
-# Give monitor extra time to flush writes in high-latency environments
-sleep 1
-
-# Check if monitor is still running before waiting
-if kill -0 "$monitor_pid" 2>/dev/null; then
-    wait "$monitor_pid" 2>/dev/null || true
-fi
-
-if ! assert_versions_in_order "$TEST_TMP/runtime_versions.log" "2 3 4"; then
+# Wait for verification process to complete
+if ! wait "$VERIFY_PID"; then
     echo "❌ Runtime did not apply versions in order (2 -> 3 -> 4)"
     exit 1
 fi