use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::hash::Hash;
use std::time::Duration;
use anyhow::{Context, Result};
use serde::Deserialize;
use thiserror::Error;
const DEFAULT_COMPLEXITY_TARGET: usize = 7_000;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum LinearOperation {
Teams,
Labels,
Projects,
ProjectTeams,
ProjectMembers,
ProjectLabels,
ProjectMilestones,
Cycles,
Issues,
Comments,
Relations,
}
impl LinearOperation {
pub fn name(self) -> &'static str {
match self {
Self::Teams => "teams",
Self::Labels => "labels",
Self::Projects => "projects",
Self::ProjectTeams => "project teams",
Self::ProjectMembers => "project members",
Self::ProjectLabels => "project labels",
Self::ProjectMilestones => "project milestones",
Self::Cycles => "cycles",
Self::Issues => "issues",
Self::Comments => "comments",
Self::Relations => "issue relations",
}
}
fn environment_name(self) -> &'static str {
match self {
Self::Teams => "TEAMS",
Self::Labels => "LABELS",
Self::Projects => "PROJECTS",
Self::ProjectTeams => "PROJECT_TEAMS",
Self::ProjectMembers => "PROJECT_MEMBERS",
Self::ProjectLabels => "PROJECT_LABELS",
Self::ProjectMilestones => "PROJECT_MILESTONES",
Self::Cycles => "CYCLES",
Self::Issues => "ISSUES",
Self::Comments => "COMMENTS",
Self::Relations => "RELATIONS",
}
}
fn recommended_page_size(self) -> usize {
match self {
Self::Projects => 25,
Self::Issues => 50,
Self::ProjectMembers | Self::ProjectLabels => 50,
Self::ProjectTeams => 25,
Self::Comments | Self::Relations => 100,
Self::Teams | Self::Labels | Self::ProjectMilestones | Self::Cycles => 100,
}
}
fn estimated_complexity(self) -> (usize, usize) {
match self {
Self::Projects => (100, 220),
Self::Issues => (100, 115),
Self::ProjectMembers | Self::ProjectLabels => (50, 80),
Self::ProjectTeams => (50, 120),
Self::Comments => (50, 45),
Self::Relations => (50, 55),
Self::ProjectMilestones => (50, 60),
Self::Cycles => (50, 45),
Self::Labels => (50, 35),
Self::Teams => (50, 30),
}
}
fn field_set(self) -> &'static str {
match self {
Self::Teams => "team identity fields",
Self::Labels => "label identity, color, and parent",
Self::Projects => "project scalar metadata, status, and lead",
Self::ProjectTeams => "project team identity fields",
Self::ProjectMembers => "project member identity fields",
Self::ProjectLabels => "project label metadata",
Self::ProjectMilestones => "milestone scalar metadata and project reference",
Self::Cycles => "cycle scalar metadata and team reference",
Self::Issues => {
"issue scalar metadata, labels, project, milestone, and cycle references"
}
Self::Comments => "comment body, author, timestamps, parent, and URL",
Self::Relations => "relation type and related issue identity",
}
}
}
#[derive(Debug, Clone)]
pub struct SyncQueryConfig {
pub complexity_target: usize,
pub minimum_page_size: usize,
pub max_retry_attempts: usize,
pub retry_base_delay: Duration,
pub verbose: bool,
page_size_overrides: HashMap<LinearOperation, usize>,
}
impl Default for SyncQueryConfig {
fn default() -> Self {
Self {
complexity_target: DEFAULT_COMPLEXITY_TARGET,
minimum_page_size: 1,
max_retry_attempts: 3,
retry_base_delay: Duration::from_millis(250),
verbose: false,
page_size_overrides: HashMap::new(),
}
}
}
impl SyncQueryConfig {
pub fn from_environment() -> Self {
let mut config = Self::default();
if let Some(value) = env_usize("RECTILINEAR_LINEAR_COMPLEXITY_TARGET") {
config.complexity_target = value.clamp(100, 9_000);
}
if let Some(value) = env_usize("RECTILINEAR_LINEAR_MIN_PAGE_SIZE") {
config.minimum_page_size = value.max(1);
}
config.verbose = std::env::var("RECTILINEAR_LINEAR_VERBOSE")
.is_ok_and(|value| matches!(value.as_str(), "1" | "true" | "yes"));
for operation in [
LinearOperation::Teams,
LinearOperation::Labels,
LinearOperation::Projects,
LinearOperation::ProjectTeams,
LinearOperation::ProjectMembers,
LinearOperation::ProjectLabels,
LinearOperation::ProjectMilestones,
LinearOperation::Cycles,
LinearOperation::Issues,
LinearOperation::Comments,
LinearOperation::Relations,
] {
let key = format!(
"RECTILINEAR_LINEAR_{}_PAGE_SIZE",
operation.environment_name()
);
if let Some(value) = env_usize(&key) {
config.page_size_overrides.insert(operation, value.max(1));
}
}
config
}
pub fn with_page_size(mut self, operation: LinearOperation, page_size: usize) -> Self {
self.page_size_overrides.insert(operation, page_size.max(1));
self
}
pub fn page_size(&self, operation: LinearOperation) -> usize {
if let Some(page_size) = self.page_size_overrides.get(&operation) {
return (*page_size).max(self.minimum_page_size);
}
let (base_cost, per_node_cost) = operation.estimated_complexity();
let planned = self
.complexity_target
.saturating_sub(base_cost)
.checked_div(per_node_cost)
.unwrap_or(1)
.max(1);
operation
.recommended_page_size()
.min(planned)
.max(self.minimum_page_size)
}
pub fn estimated_request_complexity(
&self,
operation: LinearOperation,
page_size: usize,
) -> usize {
let (base_cost, per_node_cost) = operation.estimated_complexity();
base_cost.saturating_add(per_node_cost.saturating_mul(page_size))
}
}
fn env_usize(key: &str) -> Option<usize> {
std::env::var(key).ok()?.parse().ok()
}
#[derive(Debug, Clone, Deserialize)]
pub(crate) struct PageInfo {
#[serde(rename = "hasNextPage")]
pub(crate) has_next_page: bool,
#[serde(rename = "endCursor")]
pub(crate) end_cursor: Option<String>,
}
#[derive(Debug)]
pub(crate) struct ConnectionPage<T> {
pub(crate) nodes: Vec<T>,
pub(crate) page_info: PageInfo,
}
#[derive(Debug, Clone)]
pub(crate) struct PageRequest {
pub(crate) cursor: Option<String>,
pub(crate) page_size: usize,
pub(crate) page_number: usize,
}
#[derive(Debug, Clone)]
pub(crate) struct PageContext {
pub(crate) page_size: usize,
pub(crate) cursor: Option<String>,
}
#[derive(Debug, Clone)]
pub struct SyncEvent {
pub operation: &'static str,
pub parent: Option<String>,
pub page_number: usize,
pub nodes_received: usize,
pub page_size: usize,
pub adaptive_reduction: bool,
pub completed: bool,
pub failure: Option<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct PaginationStats {
pub(crate) pages: usize,
pub(crate) nodes: usize,
pub(crate) adaptive_reductions: usize,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LinearErrorKind {
Authentication,
RateLimit,
Complexity,
Validation,
Transport,
Transient,
Api,
}
#[derive(Debug, Error)]
#[error("Linear {kind:?} error during {operation}{cursor_context}: {message}")]
pub struct LinearOperationError {
pub kind: LinearErrorKind,
pub operation: String,
pub cursor: Option<String>,
pub message: String,
pub retry_after: Option<Duration>,
cursor_context: String,
}
impl LinearOperationError {
pub fn new(
kind: LinearErrorKind,
operation: impl Into<String>,
cursor: Option<&str>,
message: impl Into<String>,
) -> Self {
let cursor = cursor.map(ToString::to_string);
let cursor_context = cursor
.as_deref()
.map(|value| format!(" at cursor '{value}'"))
.unwrap_or_default();
Self {
kind,
operation: operation.into(),
cursor,
message: message.into(),
retry_after: None,
cursor_context,
}
}
pub fn with_retry_after(mut self, retry_after: Option<Duration>) -> Self {
self.retry_after = retry_after;
self
}
}
pub(crate) fn operation_error(error: &anyhow::Error) -> Option<&LinearOperationError> {
error.downcast_ref::<LinearOperationError>()
}
pub(crate) async fn paginate<T, K, Fetch, FetchFuture, Persist, PersistFuture, KeyFn, Observe>(
config: &SyncQueryConfig,
operation: LinearOperation,
parent: Option<String>,
mut fetch: Fetch,
mut persist: Persist,
key_of: KeyFn,
mut observe: Observe,
) -> Result<PaginationStats>
where
K: Eq + Hash,
Fetch: FnMut(PageRequest) -> FetchFuture,
FetchFuture: Future<Output = Result<ConnectionPage<T>>>,
Persist: FnMut(Vec<T>, PageContext) -> PersistFuture,
PersistFuture: Future<Output = Result<()>>,
KeyFn: Fn(&T) -> K,
Observe: FnMut(SyncEvent),
{
let mut cursor = None;
let mut page_size = config.page_size(operation);
let mut stats = PaginationStats::default();
let mut retry_attempts = 0;
let mut seen = HashSet::new();
loop {
let request = PageRequest {
cursor: cursor.clone(),
page_size,
page_number: stats.pages + 1,
};
let page = match fetch(request.clone()).await {
Ok(page) => {
retry_attempts = 0;
page
}
Err(error) => {
let classified = operation_error(&error);
if classified.is_some_and(|value| value.kind == LinearErrorKind::Complexity) {
if page_size > config.minimum_page_size {
page_size = (page_size / 2).max(config.minimum_page_size);
stats.adaptive_reductions += 1;
observe(SyncEvent {
operation: operation.name(),
parent: parent.clone(),
page_number: request.page_number,
nodes_received: 0,
page_size,
adaptive_reduction: true,
completed: false,
failure: None,
});
continue;
}
let diagnostic = if page_size == 1 {
format!(
"Linear rejected a one-node {} request as too complex at cursor {:?}; \
split the operation or reduce the requested field set ({})",
operation.name(),
cursor,
operation.field_set()
)
} else {
format!(
"Linear rejected the minimum configured page size ({page_size}) for {} \
at cursor {:?}; lower RECTILINEAR_LINEAR_MIN_PAGE_SIZE or split the \
requested field set ({})",
operation.name(),
cursor,
operation.field_set()
)
};
observe(SyncEvent {
operation: operation.name(),
parent: parent.clone(),
page_number: request.page_number,
nodes_received: 0,
page_size,
adaptive_reduction: stats.adaptive_reductions > 0,
completed: false,
failure: Some(diagnostic.clone()),
});
anyhow::bail!(diagnostic);
}
let retry_delay = classified.and_then(|value| match value.kind {
LinearErrorKind::RateLimit
| LinearErrorKind::Transport
| LinearErrorKind::Transient
if retry_attempts < config.max_retry_attempts =>
{
Some(value.retry_after.unwrap_or_else(|| {
config
.retry_base_delay
.saturating_mul(1_u32 << retry_attempts.min(10))
}))
}
_ => None,
});
if let Some(delay) = retry_delay {
retry_attempts += 1;
observe(SyncEvent {
operation: operation.name(),
parent: parent.clone(),
page_number: request.page_number,
nodes_received: 0,
page_size,
adaptive_reduction: stats.adaptive_reductions > 0,
completed: false,
failure: Some(format!(
"retrying attempt {retry_attempts}/{} after {delay:?} following a {:?} error",
config.max_retry_attempts,
classified.expect("retryable errors are classified").kind
)),
});
if !delay.is_zero() {
tokio::time::sleep(delay).await;
}
continue;
}
let failure = format!("{error:#}");
observe(SyncEvent {
operation: operation.name(),
parent: parent.clone(),
page_number: request.page_number,
nodes_received: 0,
page_size,
adaptive_reduction: stats.adaptive_reductions > 0,
completed: false,
failure: Some(failure),
});
return Err(error).with_context(|| {
format!(
"failed to paginate {} at cursor {:?}",
operation.name(),
cursor
)
});
}
};
if page.page_info.has_next_page && page.page_info.end_cursor.is_none() {
anyhow::bail!(
"Malformed {} pagination response on page {}: hasNextPage was true but endCursor was missing",
operation.name(),
request.page_number
);
}
if page.page_info.has_next_page && page.page_info.end_cursor == cursor {
anyhow::bail!(
"Malformed {} pagination response on page {}: endCursor did not advance",
operation.name(),
request.page_number
);
}
let mut nodes = page.nodes;
nodes.retain(|node| seen.insert(key_of(node)));
let received = nodes.len();
persist(
nodes,
PageContext {
page_size,
cursor: cursor.clone(),
},
)
.await
.with_context(|| {
format!(
"failed to persist {} page {} at cursor {:?}",
operation.name(),
request.page_number,
cursor
)
})?;
stats.pages += 1;
stats.nodes += received;
let completed = !page.page_info.has_next_page;
observe(SyncEvent {
operation: operation.name(),
parent: parent.clone(),
page_number: request.page_number,
nodes_received: received,
page_size,
adaptive_reduction: stats.adaptive_reductions > 0,
completed,
failure: None,
});
if completed {
return Ok(stats);
}
cursor = page.page_info.end_cursor;
}
}
#[cfg(test)]
mod tests {
use std::collections::VecDeque;
use std::future::ready;
use super::*;
fn page(
nodes: &[usize],
has_next_page: bool,
end_cursor: Option<&str>,
) -> ConnectionPage<usize> {
ConnectionPage {
nodes: nodes.to_vec(),
page_info: PageInfo {
has_next_page,
end_cursor: end_cursor.map(ToString::to_string),
},
}
}
fn run<F>(future: F) -> F::Output
where
F: Future,
{
tokio::runtime::Builder::new_current_thread()
.enable_time()
.build()
.unwrap()
.block_on(future)
}
#[test]
fn single_page_connection_yields_each_node() {
let mut persisted = Vec::new();
let stats = run(paginate(
&SyncQueryConfig::default(),
LinearOperation::Issues,
None,
|_| ready(Ok(page(&[1, 2], false, None))),
|nodes, _| {
persisted.extend(nodes);
ready(Ok(()))
},
|node| *node,
|_| {},
))
.unwrap();
assert_eq!(persisted, [1, 2]);
assert_eq!(stats.pages, 1);
}
#[test]
fn multiple_pages_traverse_cursors_and_remove_boundary_duplicates() {
let mut responses = VecDeque::from([
page(&[1, 2], true, Some("cursor-1")),
page(&[2, 3], false, None),
]);
let mut requested = Vec::new();
let mut persisted = Vec::new();
let stats = run(paginate(
&SyncQueryConfig::default(),
LinearOperation::Issues,
None,
|request| {
requested.push(request.cursor);
ready(Ok(responses.pop_front().unwrap()))
},
|nodes, _| {
persisted.extend(nodes);
ready(Ok(()))
},
|node| *node,
|_| {},
))
.unwrap();
assert_eq!(requested, [None, Some("cursor-1".into())]);
assert_eq!(persisted, [1, 2, 3]);
assert_eq!(stats.nodes, 3);
}
#[test]
fn empty_connection_persists_an_empty_page() {
let mut pages = 0;
let stats = run(paginate(
&SyncQueryConfig::default(),
LinearOperation::Comments,
None,
|_| ready(Ok(page(&[], false, None))),
|nodes, _| {
assert!(nodes.is_empty());
pages += 1;
ready(Ok(()))
},
|node| *node,
|_| {},
))
.unwrap();
assert_eq!(pages, 1);
assert_eq!(stats.nodes, 0);
}
#[test]
fn malformed_pagination_metadata_is_rejected() {
let error = run(paginate(
&SyncQueryConfig::default(),
LinearOperation::Comments,
None,
|_| ready(Ok(page(&[1], true, None))),
|_, _| ready(Ok(())),
|node| *node,
|_| {},
))
.unwrap_err();
assert!(error.to_string().contains("endCursor was missing"));
}
#[test]
fn transient_transport_failure_retries_same_cursor() {
let mut attempts = 0;
let mut events = Vec::new();
let config = SyncQueryConfig {
retry_base_delay: Duration::ZERO,
..Default::default()
};
let stats = run(paginate(
&config,
LinearOperation::Issues,
None,
|request| {
attempts += 1;
if attempts == 1 {
ready(Err(LinearOperationError::new(
LinearErrorKind::Transport,
"issues",
request.cursor.as_deref(),
"connection reset",
)
.into()))
} else {
ready(Ok(page(&[1], false, None)))
}
},
|_, _| ready(Ok(())),
|node| *node,
|event| events.push(event),
))
.unwrap();
assert_eq!(attempts, 2);
assert_eq!(stats.nodes, 1);
assert!(events.iter().any(|event| {
event
.failure
.as_deref()
.is_some_and(|failure| failure.starts_with("retrying attempt 1/"))
}));
}
#[test]
fn rate_limit_retries_without_advancing_cursor() {
let mut cursors = Vec::new();
let config = SyncQueryConfig {
retry_base_delay: Duration::ZERO,
..Default::default()
};
run(paginate(
&config,
LinearOperation::Comments,
None,
|request| {
cursors.push(request.cursor.clone());
if cursors.len() == 1 {
ready(Err(LinearOperationError::new(
LinearErrorKind::RateLimit,
"comments",
request.cursor.as_deref(),
"too many requests",
)
.into()))
} else {
ready(Ok(page(&[], false, None)))
}
},
|_, _| ready(Ok(())),
|node| *node,
|_| {},
))
.unwrap();
assert_eq!(cursors, [None, None]);
}
#[test]
fn complexity_rejection_reduces_page_size_at_same_cursor() {
let config = SyncQueryConfig::default().with_page_size(LinearOperation::Issues, 40);
let mut sizes = Vec::new();
let stats = run(paginate(
&config,
LinearOperation::Issues,
None,
|request| {
sizes.push(request.page_size);
if request.page_size > 10 {
ready(Err(LinearOperationError::new(
LinearErrorKind::Complexity,
"issues",
request.cursor.as_deref(),
"Query complexity exceeds maximum allowed complexity",
)
.into()))
} else {
ready(Ok(page(&[1], false, None)))
}
},
|_, _| ready(Ok(())),
|node| *node,
|_| {},
))
.unwrap();
assert_eq!(sizes, [40, 20, 10]);
assert_eq!(stats.adaptive_reductions, 2);
}
#[test]
fn repeated_complexity_rejection_at_minimum_is_actionable() {
let config = SyncQueryConfig::default().with_page_size(LinearOperation::Projects, 4);
let error = run(paginate::<usize, usize, _, _, _, _, _, _>(
&config,
LinearOperation::Projects,
None,
|request| {
ready(Err(LinearOperationError::new(
LinearErrorKind::Complexity,
"projects",
request.cursor.as_deref(),
"too complex",
)
.into()))
},
|_, _| ready(Ok(())),
|node| *node,
|_| {},
))
.unwrap_err();
assert!(error.to_string().contains("one-node projects request"));
assert!(error.to_string().contains("field set"));
}
#[test]
fn repeated_complexity_rejection_stops_at_configured_minimum() {
let mut sizes = Vec::new();
let config = SyncQueryConfig {
minimum_page_size: 5,
..SyncQueryConfig::default().with_page_size(LinearOperation::Issues, 20)
};
let error = run(paginate::<usize, usize, _, _, _, _, _, _>(
&config,
LinearOperation::Issues,
None,
|request| {
sizes.push(request.page_size);
ready(Err(LinearOperationError::new(
LinearErrorKind::Complexity,
"issues",
request.cursor.as_deref(),
"too complex",
)
.into()))
},
|_, _| ready(Ok(())),
|node| *node,
|_| {},
))
.unwrap_err();
assert_eq!(sizes, [20, 10, 5]);
assert!(error
.to_string()
.contains("minimum configured page size (5)"));
}
#[test]
fn simulated_large_workspace_stays_under_target() {
let config = SyncQueryConfig::default();
let old_workspace_query_complexity = 72_400;
assert!(old_workspace_query_complexity > 10_000);
let issue_count: usize = 1_200;
let issue_page_size = config.page_size(LinearOperation::Issues);
let issue_request_count = issue_count.div_ceil(issue_page_size);
assert_eq!(issue_page_size, 50);
assert_eq!(issue_request_count, 24);
for operation in [
LinearOperation::Teams,
LinearOperation::Labels,
LinearOperation::Projects,
LinearOperation::ProjectTeams,
LinearOperation::ProjectMembers,
LinearOperation::ProjectLabels,
LinearOperation::ProjectMilestones,
LinearOperation::Cycles,
LinearOperation::Issues,
LinearOperation::Comments,
LinearOperation::Relations,
] {
let size = config.page_size(operation);
assert!(
config.estimated_request_complexity(operation, size) <= config.complexity_target,
"{} planned above target",
operation.name()
);
}
}
}