Skip to content

Require safe PostHog pagination cursors #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nicosuave merged 1 commit into from
Jul 18, 2026
Merged

Require safe PostHog pagination cursors #19

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 56 additions & 12 deletions src/importer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1895,12 +1895,10 @@ impl Importer {

let count = rows.len();
let mut events = Vec::with_capacity(count);
let mut next_cursor = cursor.clone();
let mut next_cursor = None;
for row in rows {
let row = EventRow::from_value(row)?;
if let Some(row_cursor) = row.cursor() {
next_cursor = Some(row_cursor);
}
next_cursor = row.cursor();
if let Some(event) = self.row_to_pipeline_event(row)? {
events.push(ImportBatchItem {
key: ImportKey::event(event.uuid.clone()),
Expand All @@ -1909,23 +1907,20 @@ impl Importer {
}
}

let next_cursor = validated_event_page_cursor(count, limit, next_cursor)?;

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Don’t require cursors for max-events terminal pages

When IMPORT_MAX_EVENTS makes next_limit return the exact remaining cap, a response with count == limit can be the terminal page because the next loop would break before issuing another PostHog query. This validation runs before sending and treats that capped page as needing a pagination cursor, so capped smoke-test imports fail if the Nth row lacks a timestamp/UUID even though no further pagination is needed and partial final pages are otherwise allowed.

Useful? React with 👍 / 👎.

@nicosuave nicosuave Jul 18, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed in #20. Cursor validation now runs only when the page is full and next_limit() confirms another query will be issued after send() updates the imported-event count. A full page that exactly satisfies IMPORT_MAX_EVENTS is terminal and does not require a cursor; continuing pages still fail closed without the UUID tie-breaker.


let previous_events = self.summary.events;
let sent = self.send(events).await?;
self.summary.events += sent;
if previous_events / PROGRESS_INTERVAL < self.summary.events / PROGRESS_INTERVAL {
eprintln!("PostHog import progress: events={}", self.summary.events);
}

if next_cursor.is_some() {
cursor = next_cursor;
offset = 0;
} else {
cursor = None;
offset += count;
}
if count < limit {
break;
}
cursor = next_cursor;
offset = 0;
}

Ok(())
Expand Down Expand Up @@ -2587,6 +2582,20 @@ fn events_query(
)
}

fn validated_event_page_cursor(
count: usize,
limit: usize,
cursor: Option<EventCursor>,
) -> Result<Option<EventCursor>, ImportError> {
if count == limit && cursor.is_none() {
return Err(ImportError::InvalidPostHogResponse(
"cannot safely paginate a full events page because its final row is missing a timestamp or UUID tie-breaker"
.to_string(),
));
}
Ok(cursor)
}

fn events_by_uuids_query(uuids: &[String]) -> String {
let uuid_list = uuids
.iter()
Expand Down Expand Up @@ -2918,9 +2927,10 @@ impl EventRow {
}

fn cursor(&self) -> Option<EventCursor> {
let uuid = self.uuid.as_ref().filter(|uuid| !uuid.is_empty())?.clone();
Some(EventCursor {
timestamp: self.timestamp?,
uuid: self.uuid.clone(),
uuid: Some(uuid),
})
}
}
Expand Down Expand Up @@ -3090,6 +3100,40 @@ mod tests {
assert!(!query.contains(" offset "));
}

#[test]
fn event_cursor_requires_uuid_tie_breaker() {
let timestamp = parse_datetime("2025-01-01T12:00:00Z").unwrap();
let without_uuid = EventRow {
uuid: None,
event: Some("test".to_string()),
distinct_id: Some("user-1".to_string()),
timestamp: Some(timestamp),
created_at: Some(timestamp),
properties: None,
};
let with_uuid = EventRow {
uuid: Some("event-1".to_string()),
event: Some("test".to_string()),
distinct_id: Some("user-1".to_string()),
timestamp: Some(timestamp),
created_at: Some(timestamp),
properties: None,
};

assert!(without_uuid.cursor().is_none());
assert_eq!(with_uuid.cursor().unwrap().uuid.as_deref(), Some("event-1"));
}

#[test]
fn full_events_page_requires_safe_cursor() {
let error = validated_event_page_cursor(100, 100, None).unwrap_err();

assert!(matches!(error, ImportError::InvalidPostHogResponse(_)));
assert!(validated_event_page_cursor(99, 100, None)
.unwrap()
.is_none());
}

#[test]
fn parses_import_caps_from_args() {
let config = ImportConfig::from_env_and_args([
Expand Down
Loading