+4
-2
src/bin/quickdid.rs
+4
-2
src/bin/quickdid.rs
···
17
metrics::create_metrics_publisher,
18
queue::{
19
HandleResolutionWork, QueueAdapter, create_mpsc_queue_from_channel, create_noop_queue,
20
-
create_redis_queue, create_redis_queue_with_dedup, create_sqlite_queue,
21
create_sqlite_queue_with_max_size,
22
},
23
sqlite_schema::create_sqlite_pool,
···
117
" QUEUE_REDIS_PREFIX Redis key prefix for queues (default: queue:handleresolver:)"
118
);
119
println!(" QUEUE_REDIS_TIMEOUT Queue blocking timeout in seconds (default: 5)");
120
-
println!(" QUEUE_REDIS_DEDUP_ENABLED Enable queue deduplication (default: false)");
121
println!(" QUEUE_REDIS_DEDUP_TTL TTL for dedup keys in seconds (default: 60)");
122
println!(" QUEUE_WORKER_ID Worker ID for Redis queue (default: worker1)");
123
println!(" QUEUE_BUFFER_SIZE Buffer size for MPSC queue (default: 1000)");
···
17
metrics::create_metrics_publisher,
18
queue::{
19
HandleResolutionWork, QueueAdapter, create_mpsc_queue_from_channel, create_noop_queue,
20
+
create_redis_queue, create_redis_queue_with_dedup, create_sqlite_queue,
21
create_sqlite_queue_with_max_size,
22
},
23
sqlite_schema::create_sqlite_pool,
···
117
" QUEUE_REDIS_PREFIX Redis key prefix for queues (default: queue:handleresolver:)"
118
);
119
println!(" QUEUE_REDIS_TIMEOUT Queue blocking timeout in seconds (default: 5)");
120
+
println!(
121
+
" QUEUE_REDIS_DEDUP_ENABLED Enable queue deduplication (default: false)"
122
+
);
123
println!(" QUEUE_REDIS_DEDUP_TTL TTL for dedup keys in seconds (default: 60)");
124
println!(" QUEUE_WORKER_ID Worker ID for Redis queue (default: worker1)");
125
println!(" QUEUE_BUFFER_SIZE Buffer size for MPSC queue (default: 1000)");
+6
-2
src/handle_resolver/proactive_refresh.rs
+6
-2
src/handle_resolver/proactive_refresh.rs
···
93
resolve_time_us = resolve_time,
94
"Fast resolution detected, considering proactive refresh"
95
);
96
-
97
if let Some(metrics) = &self.metrics {
98
metrics.incr("proactive_refresh.cache_hit_detected").await;
99
}
···
171
threshold: f64,
172
) -> Arc<dyn HandleResolver> {
173
Arc::new(DynProactiveRefreshResolver::with_metrics(
174
-
inner, queue, Some(metrics), cache_ttl, threshold,
175
))
176
}
177
···
93
resolve_time_us = resolve_time,
94
"Fast resolution detected, considering proactive refresh"
95
);
96
+
97
if let Some(metrics) = &self.metrics {
98
metrics.incr("proactive_refresh.cache_hit_detected").await;
99
}
···
171
threshold: f64,
172
) -> Arc<dyn HandleResolver> {
173
Arc::new(DynProactiveRefreshResolver::with_metrics(
174
+
inner,
175
+
queue,
176
+
Some(metrics),
177
+
cache_ttl,
178
+
threshold,
179
))
180
}
181
+8
-6
src/handle_resolver_task.rs
+8
-6
src/handle_resolver_task.rs
···
117
118
/// Check if an error represents a soft failure (handle not found)
119
/// rather than a real error condition.
120
-
///
121
/// These atproto_identity library errors indicate the handle doesn't support
122
/// the specific resolution method, which is normal and expected:
123
/// - error-atproto-identity-resolve-4: DNS resolution failed (no records)
···
130
error_str.contains("NoRecordsFound")
131
} else if error_str.starts_with("error-atproto-identity-resolve-5") {
132
// HTTP resolution - check if it's a hostname lookup failure
133
-
error_str.contains("No address associated with hostname") ||
134
-
error_str.contains("failed to lookup address information")
135
} else {
136
false
137
}
···
181
}
182
Ok(Err(e)) => {
183
let error_str = e.to_string();
184
-
185
if Self::is_soft_failure(&error_str) {
186
// This is a soft failure - handle simply doesn't support this resolution method
187
// Publish not-found metrics
···
388
assert!(!HandleResolverTask::is_soft_failure(dns_real_error));
389
390
// Test HTTP error that is NOT a soft failure (connection timeout)
391
-
let http_timeout = "error-atproto-identity-resolve-5 HTTP resolution failed: connection timeout";
392
assert!(!HandleResolverTask::is_soft_failure(http_timeout));
393
394
// Test HTTP error that is NOT a soft failure (500 error)
···
396
assert!(!HandleResolverTask::is_soft_failure(http_500));
397
398
// Test QuickDID errors should never be soft failures
399
-
let quickdid_error = "error-quickdid-resolve-1 Failed to resolve subject: internal server error";
400
assert!(!HandleResolverTask::is_soft_failure(quickdid_error));
401
402
// Test other atproto_identity error codes should not be soft failures
···
117
118
/// Check if an error represents a soft failure (handle not found)
119
/// rather than a real error condition.
120
+
///
121
/// These atproto_identity library errors indicate the handle doesn't support
122
/// the specific resolution method, which is normal and expected:
123
/// - error-atproto-identity-resolve-4: DNS resolution failed (no records)
···
130
error_str.contains("NoRecordsFound")
131
} else if error_str.starts_with("error-atproto-identity-resolve-5") {
132
// HTTP resolution - check if it's a hostname lookup failure
133
+
error_str.contains("No address associated with hostname")
134
+
|| error_str.contains("failed to lookup address information")
135
} else {
136
false
137
}
···
181
}
182
Ok(Err(e)) => {
183
let error_str = e.to_string();
184
+
185
if Self::is_soft_failure(&error_str) {
186
// This is a soft failure - handle simply doesn't support this resolution method
187
// Publish not-found metrics
···
388
assert!(!HandleResolverTask::is_soft_failure(dns_real_error));
389
390
// Test HTTP error that is NOT a soft failure (connection timeout)
391
+
let http_timeout =
392
+
"error-atproto-identity-resolve-5 HTTP resolution failed: connection timeout";
393
assert!(!HandleResolverTask::is_soft_failure(http_timeout));
394
395
// Test HTTP error that is NOT a soft failure (500 error)
···
397
assert!(!HandleResolverTask::is_soft_failure(http_500));
398
399
// Test QuickDID errors should never be soft failures
400
+
let quickdid_error =
401
+
"error-quickdid-resolve-1 Failed to resolve subject: internal server error";
402
assert!(!HandleResolverTask::is_soft_failure(quickdid_error));
403
404
// Test other atproto_identity error codes should not be soft failures
+4
-1
src/metrics.rs
+4
-1
src/metrics.rs
+58
-27
src/queue/redis.rs
+58
-27
src/queue/redis.rs
···
193
}
194
195
let dedup_key = self.dedup_key(item_id);
196
-
197
// Use SET NX EX to atomically set if not exists with expiry
198
// Returns OK if the key was set, Nil if it already existed
199
let result: Option<String> = deadpool_redis::redis::cmd("SET")
···
278
if self.dedup_enabled {
279
let dedup_id = work.dedup_key();
280
let is_new = self.check_and_mark_dedup(&mut conn, &dedup_id).await?;
281
-
282
if !is_new {
283
debug!(
284
dedup_key = %dedup_id,
···
539
.unwrap()
540
.as_nanos()
541
);
542
-
543
// Create adapter with deduplication enabled
544
let adapter = RedisQueueAdapter::<HandleResolutionWork>::with_dedup(
545
pool.clone(),
546
"test-worker-dedup".to_string(),
547
test_prefix.clone(),
548
1,
549
-
true, // Enable deduplication
550
-
2, // 2 second TTL for quick testing
551
);
552
553
let work = HandleResolutionWork::new("alice.example.com".to_string());
554
555
// First push should succeed
556
-
adapter.push(work.clone()).await.expect("First push should succeed");
557
-
558
// Second push of same item should be deduplicated (but still return Ok)
559
-
adapter.push(work.clone()).await.expect("Second push should succeed (deduplicated)");
560
-
561
// Queue should only have one item
562
let depth = adapter.depth().await;
563
-
assert_eq!(depth, Some(1), "Queue should only have one item after deduplication");
564
-
565
// Pull the item
566
let pulled = adapter.pull().await;
567
assert_eq!(pulled, Some(work.clone()));
568
-
569
// Queue should now be empty
570
let depth = adapter.depth().await;
571
assert_eq!(depth, Some(0), "Queue should be empty after pulling");
572
-
573
// Wait for dedup TTL to expire
574
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
575
-
576
// Should be able to push again after TTL expires
577
-
adapter.push(work.clone()).await.expect("Push after TTL expiry should succeed");
578
-
579
let depth = adapter.depth().await;
580
-
assert_eq!(depth, Some(1), "Queue should have one item after TTL expiry");
581
}
582
583
#[tokio::test]
···
599
.unwrap()
600
.as_nanos()
601
);
602
-
603
// Create adapter with deduplication disabled
604
let adapter = RedisQueueAdapter::<HandleResolutionWork>::with_dedup(
605
pool.clone(),
606
"test-worker-nodedup".to_string(),
607
test_prefix.clone(),
608
1,
609
-
false, // Disable deduplication
610
60,
611
);
612
613
let work = HandleResolutionWork::new("bob.example.com".to_string());
614
615
// Push same item twice
616
-
adapter.push(work.clone()).await.expect("First push should succeed");
617
-
adapter.push(work.clone()).await.expect("Second push should succeed");
618
-
619
// Queue should have two items (no deduplication)
620
let depth = adapter.depth().await;
621
-
assert_eq!(depth, Some(2), "Queue should have two items when deduplication is disabled");
622
-
623
// Pull both items
624
let pulled1 = adapter.pull().await;
625
assert_eq!(pulled1, Some(work.clone()));
626
-
627
let pulled2 = adapter.pull().await;
628
assert_eq!(pulled2, Some(work.clone()));
629
-
630
// Queue should now be empty
631
let depth = adapter.depth().await;
632
-
assert_eq!(depth, Some(0), "Queue should be empty after pulling all items");
633
}
634
635
#[tokio::test]
···
193
}
194
195
let dedup_key = self.dedup_key(item_id);
196
+
197
// Use SET NX EX to atomically set if not exists with expiry
198
// Returns OK if the key was set, Nil if it already existed
199
let result: Option<String> = deadpool_redis::redis::cmd("SET")
···
278
if self.dedup_enabled {
279
let dedup_id = work.dedup_key();
280
let is_new = self.check_and_mark_dedup(&mut conn, &dedup_id).await?;
281
+
282
if !is_new {
283
debug!(
284
dedup_key = %dedup_id,
···
539
.unwrap()
540
.as_nanos()
541
);
542
+
543
// Create adapter with deduplication enabled
544
let adapter = RedisQueueAdapter::<HandleResolutionWork>::with_dedup(
545
pool.clone(),
546
"test-worker-dedup".to_string(),
547
test_prefix.clone(),
548
1,
549
+
true, // Enable deduplication
550
+
2, // 2 second TTL for quick testing
551
);
552
553
let work = HandleResolutionWork::new("alice.example.com".to_string());
554
555
// First push should succeed
556
+
adapter
557
+
.push(work.clone())
558
+
.await
559
+
.expect("First push should succeed");
560
+
561
// Second push of same item should be deduplicated (but still return Ok)
562
+
adapter
563
+
.push(work.clone())
564
+
.await
565
+
.expect("Second push should succeed (deduplicated)");
566
+
567
// Queue should only have one item
568
let depth = adapter.depth().await;
569
+
assert_eq!(
570
+
depth,
571
+
Some(1),
572
+
"Queue should only have one item after deduplication"
573
+
);
574
+
575
// Pull the item
576
let pulled = adapter.pull().await;
577
assert_eq!(pulled, Some(work.clone()));
578
+
579
// Queue should now be empty
580
let depth = adapter.depth().await;
581
assert_eq!(depth, Some(0), "Queue should be empty after pulling");
582
+
583
// Wait for dedup TTL to expire
584
tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
585
+
586
// Should be able to push again after TTL expires
587
+
adapter
588
+
.push(work.clone())
589
+
.await
590
+
.expect("Push after TTL expiry should succeed");
591
+
592
let depth = adapter.depth().await;
593
+
assert_eq!(
594
+
depth,
595
+
Some(1),
596
+
"Queue should have one item after TTL expiry"
597
+
);
598
}
599
600
#[tokio::test]
···
616
.unwrap()
617
.as_nanos()
618
);
619
+
620
// Create adapter with deduplication disabled
621
let adapter = RedisQueueAdapter::<HandleResolutionWork>::with_dedup(
622
pool.clone(),
623
"test-worker-nodedup".to_string(),
624
test_prefix.clone(),
625
1,
626
+
false, // Disable deduplication
627
60,
628
);
629
630
let work = HandleResolutionWork::new("bob.example.com".to_string());
631
632
// Push same item twice
633
+
adapter
634
+
.push(work.clone())
635
+
.await
636
+
.expect("First push should succeed");
637
+
adapter
638
+
.push(work.clone())
639
+
.await
640
+
.expect("Second push should succeed");
641
+
642
// Queue should have two items (no deduplication)
643
let depth = adapter.depth().await;
644
+
assert_eq!(
645
+
depth,
646
+
Some(2),
647
+
"Queue should have two items when deduplication is disabled"
648
+
);
649
+
650
// Pull both items
651
let pulled1 = adapter.pull().await;
652
assert_eq!(pulled1, Some(work.clone()));
653
+
654
let pulled2 = adapter.pull().await;
655
assert_eq!(pulled2, Some(work.clone()));
656
+
657
// Queue should now be empty
658
let depth = adapter.depth().await;
659
+
assert_eq!(
660
+
depth,
661
+
Some(0),
662
+
"Queue should be empty after pulling all items"
663
+
);
664
}
665
666
#[tokio::test]
+1
-1
src/queue/work.rs
+1
-1
src/queue/work.rs
···
125
// Same handle should have same dedup key
126
assert_eq!(work1.dedup_key(), work2.dedup_key());
127
assert_eq!(work1.dedup_key(), "alice.example.com");
128
-
129
// Different handle should have different dedup key
130
assert_ne!(work1.dedup_key(), work3.dedup_key());
131
assert_eq!(work3.dedup_key(), "bob.example.com");
···
125
// Same handle should have same dedup key
126
assert_eq!(work1.dedup_key(), work2.dedup_key());
127
assert_eq!(work1.dedup_key(), "alice.example.com");
128
+
129
// Different handle should have different dedup key
130
assert_ne!(work1.dedup_key(), work3.dedup_key());
131
assert_eq!(work3.dedup_key(), "bob.example.com");