krabka_app_sdk/
stubs.rs

1//! Queue client plus gated stubs for modules not implemented in v1.
2
3use bytes::Bytes;
4
5use crate::{
6    client::{
7        KrabkaClient, MockMessage, MockQueueDelivery, MockQueueDisposition, MockQueueMessageState,
8        MockQueueSession,
9    },
10    error::KrabkaError,
11    pb,
12};
13
14const SUPPORTED_QUEUE_LOCK_DURATION_MS: u64 = 30_000;
15
16/// Queues module client.
17#[derive(Debug, Clone)]
18pub struct QueuesClient {
19    client: KrabkaClient,
20}
21
22impl QueuesClient {
23    pub(crate) fn new(client: KrabkaClient) -> Self {
24        Self { client }
25    }
26
27    /// Acquire queue messages.
28    ///
29    /// # Errors
30    ///
31    /// Returns an error for invalid queue options, transport failures, or a
32    /// gateway request error.
33    pub async fn acquire(
34        &self,
35        topic: impl Into<String>,
36        group: impl Into<String>,
37        options: AcquireOptions,
38    ) -> Result<QueueAcquireResult, KrabkaError> {
39        self.acquire_with_session(topic, group, options, "").await
40    }
41
42    /// Acquire queue messages, optionally reusing a gateway session.
43    ///
44    /// # Errors
45    ///
46    /// Returns an error for invalid queue options, transport failures, or a
47    /// gateway request error.
48    pub async fn acquire_with_session(
49        &self,
50        topic: impl Into<String>,
51        group: impl Into<String>,
52        options: AcquireOptions,
53        session_id: impl Into<String>,
54    ) -> Result<QueueAcquireResult, KrabkaError> {
55        let topic = topic.into();
56        let group = group.into();
57        if group.is_empty() {
58            return Err(KrabkaError::InvalidArgument(
59                "queue group is required".into(),
60            ));
61        }
62        if topic.is_empty() {
63            return Err(KrabkaError::InvalidArgument(
64                "queue topic is required".into(),
65            ));
66        }
67        if options.lock_duration_ms != SUPPORTED_QUEUE_LOCK_DURATION_MS {
68            return Err(KrabkaError::InvalidArgument(format!(
69                "queue lock_duration_ms must be {SUPPORTED_QUEUE_LOCK_DURATION_MS}; per-acquire lock durations are not supported"
70            )));
71        }
72        let session_id = session_id.into();
73
74        if self.client.is_mock() {
75            return self.acquire_mock(&topic, &group, options, &session_id);
76        }
77        if self.client.is_unreachable() {
78            return Err(KrabkaError::Transport("endpoint unreachable".into()));
79        }
80
81        let response: pb::QueueAcquireResponse = self
82            .client
83            .inner
84            .connect
85            .unary(
86                "/krabka.gateway.v1.Gateway/QueueAcquire",
87                &pb::QueueAcquireRequest {
88                    group_id: group,
89                    topics: vec![topic],
90                    max_messages: options.max,
91                    wait_ms: options.wait_ms,
92                    session_id,
93                    lock_duration_ms: options.lock_duration_ms,
94                },
95            )
96            .await?;
97        Ok(QueueAcquireResult {
98            session_id: response.session_id,
99            messages: response
100                .messages
101                .into_iter()
102                .map(QueueMessage::from)
103                .collect(),
104        })
105    }
106
107    /// Acknowledge queue messages.
108    ///
109    /// # Errors
110    ///
111    /// Returns an error when the session is invalid or the gateway request
112    /// fails. Per-entry failures are returned in the result vector.
113    pub async fn acknowledge(
114        &self,
115        session_id: impl Into<String>,
116        entries: Vec<QueueAckEntry>,
117    ) -> Result<Vec<QueueAckResult>, KrabkaError> {
118        let session_id = session_id.into();
119        if self.client.is_mock() {
120            return self.acknowledge_mock(&session_id, entries);
121        }
122        if self.client.is_unreachable() {
123            return Err(KrabkaError::Transport("endpoint unreachable".into()));
124        }
125
126        let response: pb::QueueAcknowledgeResponse = self
127            .client
128            .inner
129            .connect
130            .unary(
131                "/krabka.gateway.v1.Gateway/QueueAcknowledge",
132                &pb::QueueAcknowledgeRequest {
133                    session_id,
134                    entries: entries.into_iter().map(pb::QueueAckEntry::from).collect(),
135                },
136            )
137            .await?;
138        Ok(response
139            .results
140            .into_iter()
141            .map(QueueAckResult::from)
142            .collect())
143    }
144
145    /// Renew queue message locks.
146    ///
147    /// # Errors
148    ///
149    /// Returns an error when the session is invalid or the gateway request
150    /// fails. Per-entry failures are returned in the result vector.
151    pub async fn renew(
152        &self,
153        session_id: impl Into<String>,
154        entries: Vec<QueueRenewEntry>,
155    ) -> Result<Vec<QueueAckResult>, KrabkaError> {
156        let session_id = session_id.into();
157        if self.client.is_mock() {
158            return self.renew_mock(&session_id, entries);
159        }
160        if self.client.is_unreachable() {
161            return Err(KrabkaError::Transport("endpoint unreachable".into()));
162        }
163
164        let response: pb::QueueRenewResponse = self
165            .client
166            .inner
167            .connect
168            .unary(
169                "/krabka.gateway.v1.Gateway/QueueRenew",
170                &pb::QueueRenewRequest {
171                    session_id,
172                    entries: entries.into_iter().map(pb::QueueAckEntry::from).collect(),
173                },
174            )
175            .await?;
176        Ok(response
177            .results
178            .into_iter()
179            .map(QueueAckResult::from)
180            .collect())
181    }
182
183    fn acquire_mock(
184        &self,
185        topic: &str,
186        group: &str,
187        options: AcquireOptions,
188        requested_session_id: &str,
189    ) -> Result<QueueAcquireResult, KrabkaError> {
190        let mut state = self
191            .client
192            .inner
193            .mock
194            .lock()
195            .expect("mock state lock not poisoned");
196        let effective_max = options.max.clamp(1, 500);
197        let session_id = mock_session_id(
198            &mut state.queue_sessions,
199            requested_session_id,
200            topic,
201            group,
202            options.max,
203            effective_max,
204        )?;
205        let delivered = state
206            .messages
207            .iter()
208            .filter(|message| message.topic == topic)
209            .filter(|message| mock_message_is_available(&state.queue_states, group, message))
210            .take(usize::try_from(effective_max).unwrap_or(usize::MAX))
211            .cloned()
212            .collect::<Vec<_>>();
213
214        let messages = delivered
215            .into_iter()
216            .map(|message| {
217                let delivery = MockQueueDelivery::from(&message);
218                let queue_state = mock_queue_state_mut(&mut state.queue_states, group, delivery);
219                queue_state.disposition = MockQueueDisposition::Acquired(session_id.clone());
220                queue_state.delivery_count = queue_state.delivery_count.saturating_add(1);
221                queue_message_from_mock(message, queue_state.delivery_count)
222            })
223            .collect();
224
225        Ok(QueueAcquireResult {
226            session_id,
227            messages,
228        })
229    }
230
231    fn acknowledge_mock(
232        &self,
233        session_id: &str,
234        entries: Vec<QueueAckEntry>,
235    ) -> Result<Vec<QueueAckResult>, KrabkaError> {
236        let mut state = self
237            .client
238            .inner
239            .mock
240            .lock()
241            .expect("mock state lock not poisoned");
242        let Some(session) = state
243            .queue_sessions
244            .iter()
245            .find(|session| session.id == session_id)
246            .cloned()
247        else {
248            return Err(KrabkaError::InvalidArgument(
249                "queue session expired; re-acquire".into(),
250            ));
251        };
252
253        let mut results = Vec::with_capacity(entries.len());
254        for entry in entries {
255            let delivery = MockQueueDelivery::from(&entry);
256            let Some(queue_state) = state.queue_states.iter_mut().find(|queue_state| {
257                queue_state.group == session.group
258                    && queue_state.delivery == delivery
259                    && matches!(
260                        &queue_state.disposition,
261                        MockQueueDisposition::Acquired(owner) if owner == session_id
262                    )
263            }) else {
264                results.push(queue_result_error(
265                    entry.into(),
266                    "queue message is not acquired",
267                ));
268                continue;
269            };
270
271            queue_state.disposition = match entry.ack_type {
272                QueueAckType::Accept => MockQueueDisposition::Accepted,
273                QueueAckType::Release => MockQueueDisposition::Available,
274                QueueAckType::Reject => MockQueueDisposition::Rejected,
275            };
276            results.push(QueueAckResult {
277                entry: entry.into(),
278                error: None,
279            });
280        }
281        Ok(results)
282    }
283
284    fn renew_mock(
285        &self,
286        session_id: &str,
287        entries: Vec<QueueRenewEntry>,
288    ) -> Result<Vec<QueueAckResult>, KrabkaError> {
289        let state = self
290            .client
291            .inner
292            .mock
293            .lock()
294            .expect("mock state lock not poisoned");
295        let Some(session) = state
296            .queue_sessions
297            .iter()
298            .find(|session| session.id == session_id)
299        else {
300            return Err(KrabkaError::InvalidArgument(
301                "queue session expired; re-acquire".into(),
302            ));
303        };
304
305        Ok(entries
306            .into_iter()
307            .map(|entry| {
308                let delivery = MockQueueDelivery::from(&entry);
309                if state.queue_states.iter().any(|queue_state| {
310                    queue_state.group == session.group
311                        && queue_state.delivery == delivery
312                        && matches!(
313                            &queue_state.disposition,
314                            MockQueueDisposition::Acquired(owner) if owner == session_id
315                        )
316                }) {
317                    return QueueAckResult { entry, error: None };
318                }
319                queue_result_error(entry, "queue message is not acquired")
320            })
321            .collect())
322    }
323}
324
325/// Queue acquire options.
326#[derive(Debug, Clone, Copy, PartialEq, Eq)]
327pub struct AcquireOptions {
328    /// Maximum messages to acquire.
329    pub max: u32,
330    /// Lock duration in milliseconds.
331    pub lock_duration_ms: u64,
332    /// Maximum server wait time in milliseconds.
333    pub wait_ms: u32,
334}
335
336impl Default for AcquireOptions {
337    fn default() -> Self {
338        Self {
339            max: 1,
340            lock_duration_ms: 30_000,
341            wait_ms: 1_000,
342        }
343    }
344}
345
346/// Queue acquire result.
347#[derive(Debug, Clone, PartialEq, Eq)]
348pub struct QueueAcquireResult {
349    /// Gateway session id.
350    pub session_id: String,
351    /// Acquired messages.
352    pub messages: Vec<QueueMessage>,
353}
354
355/// Acquired queue message.
356#[derive(Debug, Clone, PartialEq, Eq)]
357pub struct QueueMessage {
358    /// Topic name.
359    pub topic: String,
360    /// Partition index.
361    pub partition: i32,
362    /// Record offset.
363    pub offset: i64,
364    /// Record value, or `None` for a Kafka tombstone.
365    pub value: Option<Bytes>,
366    /// Headers in delivery order.
367    pub headers: Vec<(String, Option<Bytes>)>,
368    /// Gateway delivery count.
369    pub delivery_count: i32,
370}
371
372impl QueueMessage {
373    /// Stable conformance message id.
374    #[must_use]
375    pub fn message_id(&self) -> String {
376        format!("{}:{}:{}", self.topic, self.partition, self.offset)
377    }
378}
379
380impl From<pb::QueuedMessage> for QueueMessage {
381    fn from(value: pb::QueuedMessage) -> Self {
382        Self {
383            topic: value.topic,
384            partition: value.partition,
385            offset: value.offset,
386            value: value.value.map(Bytes::from),
387            headers: value
388                .headers
389                .into_iter()
390                .map(|header| (header.key, header.value.map(Bytes::from)))
391                .collect(),
392            delivery_count: value.delivery_count,
393        }
394    }
395}
396
397/// Queue acknowledgement type.
398#[derive(Debug, Clone, Copy, PartialEq, Eq)]
399pub enum QueueAckType {
400    /// Accept and remove the message.
401    Accept,
402    /// Release the message for redelivery.
403    Release,
404    /// Reject the message.
405    Reject,
406}
407
408/// Queue acknowledgement entry.
409#[derive(Debug, Clone, PartialEq, Eq)]
410pub struct QueueAckEntry {
411    /// Topic name.
412    pub topic: String,
413    /// Partition index.
414    pub partition: i32,
415    /// Record offset.
416    pub offset: i64,
417    /// Ack verdict.
418    pub ack_type: QueueAckType,
419}
420
421impl From<QueueAckEntry> for pb::QueueAckEntry {
422    fn from(value: QueueAckEntry) -> Self {
423        Self {
424            topic: value.topic,
425            partition: value.partition,
426            offset: value.offset,
427            r#type: match value.ack_type {
428                QueueAckType::Accept => pb::QueueAckType::Accept,
429                QueueAckType::Release => pb::QueueAckType::Release,
430                QueueAckType::Reject => pb::QueueAckType::Reject,
431            } as i32,
432        }
433    }
434}
435
436/// Queue renew entry.
437#[derive(Debug, Clone, PartialEq, Eq)]
438pub struct QueueRenewEntry {
439    /// Topic name.
440    pub topic: String,
441    /// Partition index.
442    pub partition: i32,
443    /// Record offset.
444    pub offset: i64,
445}
446
447impl From<QueueRenewEntry> for pb::QueueAckEntry {
448    fn from(value: QueueRenewEntry) -> Self {
449        Self {
450            topic: value.topic,
451            partition: value.partition,
452            offset: value.offset,
453            r#type: pb::QueueAckType::Unspecified as i32,
454        }
455    }
456}
457
458impl From<&QueueAckEntry> for MockQueueDelivery {
459    fn from(value: &QueueAckEntry) -> Self {
460        Self {
461            topic: value.topic.clone(),
462            partition: value.partition,
463            offset: value.offset,
464        }
465    }
466}
467
468impl From<&QueueRenewEntry> for MockQueueDelivery {
469    fn from(value: &QueueRenewEntry) -> Self {
470        Self {
471            topic: value.topic.clone(),
472            partition: value.partition,
473            offset: value.offset,
474        }
475    }
476}
477
478impl From<&MockMessage> for MockQueueDelivery {
479    fn from(value: &MockMessage) -> Self {
480        Self {
481            topic: value.topic.clone(),
482            partition: value.partition,
483            offset: value.offset,
484        }
485    }
486}
487
488impl From<QueueAckEntry> for QueueRenewEntry {
489    fn from(value: QueueAckEntry) -> Self {
490        Self {
491            topic: value.topic,
492            partition: value.partition,
493            offset: value.offset,
494        }
495    }
496}
497
498/// Per-entry queue acknowledgement result.
499#[derive(Debug, PartialEq, Eq)]
500pub struct QueueAckResult {
501    /// Original entry key.
502    pub entry: QueueRenewEntry,
503    /// Optional per-entry error.
504    pub error: Option<KrabkaError>,
505}
506
507impl From<pb::QueueAckResult> for QueueAckResult {
508    fn from(value: pb::QueueAckResult) -> Self {
509        let entry = value.entry.unwrap_or_default();
510        Self {
511            entry: QueueRenewEntry {
512                topic: entry.topic,
513                partition: entry.partition,
514                offset: entry.offset,
515            },
516            error: value.error.map(error_info_to_krabka),
517        }
518    }
519}
520
521fn error_info_to_krabka(error: pb::ErrorInfo) -> KrabkaError {
522    let message = error.message;
523    if error.retriable {
524        return KrabkaError::Transport(message);
525    }
526    match error.code {
527        3 | 9 => KrabkaError::InvalidArgument(message),
528        16 => KrabkaError::Unauthenticated(message),
529        5 => KrabkaError::NotFound(message),
530        _ => KrabkaError::ServerError(message),
531    }
532}
533
534fn mock_session_id(
535    sessions: &mut Vec<MockQueueSession>,
536    requested_session_id: &str,
537    topic: &str,
538    group: &str,
539    requested_max: u32,
540    effective_max: u32,
541) -> Result<String, KrabkaError> {
542    if !requested_session_id.is_empty() {
543        let Some(session) = sessions
544            .iter()
545            .find(|session| session.id == requested_session_id)
546        else {
547            return Err(KrabkaError::InvalidArgument(
548                "queue session expired; re-acquire".into(),
549            ));
550        };
551        if session.topic != topic || session.group != group {
552            return Err(KrabkaError::InvalidArgument(
553                "group_id and topics are fixed when a queue session is created".into(),
554            ));
555        }
556        if requested_max != 0 && effective_max != session.max_messages {
557            return Err(KrabkaError::InvalidArgument(
558                "max_messages is fixed when a queue session is created".into(),
559            ));
560        }
561        return Ok(requested_session_id.to_string());
562    }
563
564    let session_id = format!("mock-queue-session-{}", sessions.len() + 1);
565    sessions.push(MockQueueSession {
566        id: session_id.clone(),
567        topic: topic.to_string(),
568        group: group.to_string(),
569        max_messages: effective_max,
570    });
571    Ok(session_id)
572}
573
574fn mock_message_is_available(
575    queue_states: &[MockQueueMessageState],
576    group: &str,
577    message: &MockMessage,
578) -> bool {
579    let delivery = MockQueueDelivery::from(message);
580    queue_states
581        .iter()
582        .find(|queue_state| queue_state.group == group && queue_state.delivery == delivery)
583        .is_none_or(|queue_state| queue_state.disposition == MockQueueDisposition::Available)
584}
585
586fn mock_queue_state_mut<'a>(
587    queue_states: &'a mut Vec<MockQueueMessageState>,
588    group: &str,
589    delivery: MockQueueDelivery,
590) -> &'a mut MockQueueMessageState {
591    if let Some(index) = queue_states
592        .iter()
593        .position(|queue_state| queue_state.group == group && queue_state.delivery == delivery)
594    {
595        return &mut queue_states[index];
596    }
597    queue_states.push(MockQueueMessageState {
598        group: group.to_string(),
599        delivery,
600        disposition: MockQueueDisposition::Available,
601        delivery_count: 0,
602    });
603    queue_states
604        .last_mut()
605        .expect("queue state was just inserted")
606}
607
608fn queue_message_from_mock(message: MockMessage, delivery_count: i32) -> QueueMessage {
609    QueueMessage {
610        topic: message.topic,
611        partition: message.partition,
612        offset: message.offset,
613        value: Some(message.value),
614        headers: message.headers,
615        delivery_count,
616    }
617}
618
619fn queue_result_error(entry: QueueRenewEntry, message: impl Into<String>) -> QueueAckResult {
620    QueueAckResult {
621        entry,
622        error: Some(KrabkaError::InvalidArgument(message.into())),
623    }
624}
625
626/// Database module client.
627#[derive(Debug, Clone, Copy, Default)]
628pub struct DatabaseClient;
629
630impl DatabaseClient {
631    pub(crate) fn new() -> Self {
632        Self
633    }
634
635    /// Connect to a named database.
636    ///
637    /// # Errors
638    ///
639    /// Always returns [`KrabkaError::Unimplemented`] while the database module
640    /// remains gated.
641    pub fn connect(&self, _name: &str) -> Result<(), KrabkaError> {
642        Err(KrabkaError::Unimplemented {
643            module: "database",
644            gated_on: "chapter-f-control-plane",
645        })
646    }
647}
648
649/// Auth module client.
650#[derive(Debug, Clone, Default)]
651pub struct AuthClient {
652    bearer: Option<String>,
653}
654
655impl AuthClient {
656    pub(crate) fn new(bearer: Option<String>) -> Self {
657        Self { bearer }
658    }
659
660    /// Return the configured bearer token, if any.
661    #[must_use]
662    pub fn bearer_token(&self) -> Option<&str> {
663        self.bearer.as_deref()
664    }
665
666    /// Sign-in is outside contract v1.
667    ///
668    /// # Errors
669    ///
670    /// Always returns [`KrabkaError::Unauthenticated`] because sign-in is not
671    /// part of contract v1.
672    pub fn sign_in(&self, _username: &str, _password: &str) -> Result<(), KrabkaError> {
673        Err(KrabkaError::Unauthenticated(
674            "identity APIs are not part of contract v1".into(),
675        ))
676    }
677}
678
679/// Blob module client.
680#[derive(Debug, Clone, Copy, Default)]
681pub struct BlobClient;
682
683impl BlobClient {
684    pub(crate) fn new() -> Self {
685        Self
686    }
687
688    /// Put a blob.
689    ///
690    /// # Errors
691    ///
692    /// Always returns [`KrabkaError::Unimplemented`] while the blob module
693    /// remains gated.
694    pub fn put(&self, _key: &str, _value: Bytes) -> Result<(), KrabkaError> {
695        Err(KrabkaError::Unimplemented {
696            module: "blob",
697            gated_on: "chapter-b-blob-api",
698        })
699    }
700
701    /// Get a blob.
702    ///
703    /// # Errors
704    ///
705    /// Always returns [`KrabkaError::Unimplemented`] while the blob module
706    /// remains gated.
707    pub fn get(&self, _key: &str) -> Result<Bytes, KrabkaError> {
708        Err(KrabkaError::Unimplemented {
709            module: "blob",
710            gated_on: "chapter-b-blob-api",
711        })
712    }
713}
714
715#[cfg(test)]
716mod tests {
717    use super::*;
718    use crate::KrabkaClient;
719
720    #[tokio::test]
721    async fn mock_queues_acquire_published_records() {
722        let client = KrabkaClient::new("mock://gateway").unwrap();
723        client
724            .messaging()
725            .publish(
726                "work",
727                Bytes::from_static(b"job"),
728                crate::PublishOptions::default(),
729            )
730            .await
731            .expect("publish to mock");
732
733        let acquired = client
734            .queues()
735            .acquire("work", "workers", AcquireOptions::default())
736            .await
737            .expect("acquire from mock");
738
739        assert_eq!(acquired.session_id, "mock-queue-session-1");
740        assert_eq!(acquired.messages.len(), 1);
741        assert_eq!(acquired.messages[0].value, Some(Bytes::from_static(b"job")));
742    }
743
744    #[test]
745    fn queue_messages_distinguish_tombstones_from_empty_values() {
746        let message = |value| {
747            QueueMessage::from(pb::QueuedMessage {
748                topic: "work".into(),
749                value,
750                ..pb::QueuedMessage::default()
751            })
752        };
753
754        assert2::assert!(message(None).value == None);
755        assert2::assert!(message(Some(Vec::new())).value == Some(Bytes::new()));
756    }
757
758    #[tokio::test]
759    async fn mock_queue_sessions_own_delivered_coordinates() {
760        let client = KrabkaClient::new("mock://gateway").unwrap();
761        client
762            .messaging()
763            .publish(
764                "work",
765                Bytes::from_static(b"job"),
766                crate::PublishOptions::default(),
767            )
768            .await
769            .expect("publish to mock");
770        let first = client
771            .queues()
772            .acquire("work", "workers", AcquireOptions::default())
773            .await
774            .expect("first session");
775        let second = client
776            .queues()
777            .acquire("work", "workers", AcquireOptions::default())
778            .await
779            .expect("second session");
780        let entry = QueueAckEntry {
781            topic: "work".into(),
782            partition: 0,
783            offset: 0,
784            ack_type: QueueAckType::Accept,
785        };
786
787        let wrong_session = client
788            .queues()
789            .acknowledge(&second.session_id, vec![entry.clone()])
790            .await
791            .expect("wrong-session acknowledgement returns per-entry result");
792        assert2::assert!(
793            wrong_session
794                == vec![queue_result_error(
795                    QueueRenewEntry::from(entry.clone()),
796                    "queue message is not acquired"
797                )]
798        );
799        let renewed = client
800            .queues()
801            .renew(
802                &second.session_id,
803                vec![QueueRenewEntry::from(entry.clone())],
804            )
805            .await
806            .expect("wrong-session renew returns per-entry result");
807        assert2::assert!(
808            renewed
809                == vec![queue_result_error(
810                    QueueRenewEntry::from(entry),
811                    "queue message is not acquired"
812                )]
813        );
814
815        let missing = client
816            .queues()
817            .acquire_with_session(
818                "work",
819                "workers",
820                AcquireOptions::default(),
821                "missing-session",
822            )
823            .await;
824        assert2::assert!(
825            missing
826                == Err(KrabkaError::InvalidArgument(
827                    "queue session expired; re-acquire".into()
828                ))
829        );
830        let changed = client
831            .queues()
832            .acquire_with_session(
833                "work",
834                "other-workers",
835                AcquireOptions::default(),
836                first.session_id,
837            )
838            .await;
839        assert2::assert!(
840            changed
841                == Err(KrabkaError::InvalidArgument(
842                    "group_id and topics are fixed when a queue session is created".into()
843                ))
844        );
845    }
846
847    #[tokio::test]
848    async fn mock_queue_delivery_state_is_independent_per_group() {
849        let client = KrabkaClient::new("mock://gateway").unwrap();
850        client
851            .messaging()
852            .publish(
853                "work",
854                Bytes::from_static(b"job"),
855                crate::PublishOptions::default(),
856            )
857            .await
858            .expect("publish to mock");
859
860        let first = client
861            .queues()
862            .acquire("work", "workers-a", AcquireOptions::default())
863            .await
864            .expect("group A acquire");
865        let second = client
866            .queues()
867            .acquire("work", "workers-b", AcquireOptions::default())
868            .await
869            .expect("group B acquire");
870
871        assert2::assert!(first.messages.len() == 1);
872        assert2::assert!(second.messages.len() == 1);
873        assert2::assert!(first.messages[0].delivery_count == 1);
874        assert2::assert!(second.messages[0].delivery_count == 1);
875
876        let coordinate = QueueAckEntry {
877            topic: "work".into(),
878            partition: 0,
879            offset: 0,
880            ack_type: QueueAckType::Accept,
881        };
882        let accepted = client
883            .queues()
884            .acknowledge(&second.session_id, vec![coordinate.clone()])
885            .await
886            .expect("group B acknowledge");
887        assert2::assert!(accepted[0].error.is_none());
888
889        let renewed = client
890            .queues()
891            .renew(
892                &first.session_id,
893                vec![QueueRenewEntry::from(coordinate.clone())],
894            )
895            .await
896            .expect("group A renew");
897        assert2::assert!(renewed[0].error.is_none());
898
899        let accepted = client
900            .queues()
901            .acknowledge(&first.session_id, vec![coordinate])
902            .await
903            .expect("group A acknowledge");
904        assert2::assert!(accepted[0].error.is_none());
905    }
906
907    #[test]
908    fn retriable_queue_error_overrides_gateway_code_and_preserves_message() {
909        assert2::assert!(
910            error_info_to_krabka(pb::ErrorInfo {
911                code: 9,
912                message: "coordinator retry".into(),
913                retriable: true,
914            }) == KrabkaError::Transport("coordinator retry".into())
915        );
916        assert2::assert!(
917            error_info_to_krabka(pb::ErrorInfo {
918                code: 9,
919                message: "record is not acquired by this session".into(),
920                retriable: false,
921            }) == KrabkaError::InvalidArgument("record is not acquired by this session".into())
922        );
923    }
924
925    #[tokio::test]
926    async fn stub_modules_return_gated_unimplemented() {
927        let client = KrabkaClient::new("mock://gateway").unwrap();
928        assert_eq!(
929            client.database().connect("orders"),
930            Err(KrabkaError::Unimplemented {
931                module: "database",
932                gated_on: "chapter-f-control-plane"
933            })
934        );
935        assert_eq!(
936            client.blob().put("k", Bytes::from_static(b"v")),
937            Err(KrabkaError::Unimplemented {
938                module: "blob",
939                gated_on: "chapter-b-blob-api"
940            })
941        );
942    }
943}
krabka_app_sdk/stubs.rs

krabka_app_sdk/
stubs.rs
