gear_core/message/
context.rs

1// This file is part of Gear.
2
3// Copyright (C) 2022-2025 Gear Technologies Inc.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19use crate::{
20    buffer::Payload,
21    ids::{ActorId, MessageId, ReservationId, prelude::*},
22    message::{
23        Dispatch, HandleMessage, HandlePacket, IncomingMessage, InitMessage, InitPacket,
24        ReplyMessage, ReplyPacket,
25    },
26    reservation::{GasReserver, ReservationNonce},
27};
28use alloc::{
29    collections::{BTreeMap, BTreeSet},
30    vec::Vec,
31};
32use gear_core_errors::{ExecutionError, ExtError, MessageError as Error, MessageError};
33use parity_scale_codec::{Decode, Encode};
34use scale_decode::DecodeAsType;
35use scale_encode::EncodeAsType;
36use scale_info::TypeInfo;
37
38use super::{DispatchKind, IncomingDispatch, Packet};
39
40/// Context settings.
41#[derive(Clone, Copy, Debug, Default)]
42pub struct ContextSettings {
43    /// Fee for sending message.
44    pub sending_fee: u64,
45    /// Fee for sending scheduled message.
46    pub scheduled_sending_fee: u64,
47    /// Fee for calling wait.
48    pub waiting_fee: u64,
49    /// Fee for waking messages.
50    pub waking_fee: u64,
51    /// Fee for creating reservation.
52    pub reservation_fee: u64,
53    /// Limit of outgoing messages, that program can send in current message processing.
54    pub outgoing_limit: u32,
55    /// Limit of bytes in outgoing messages during current execution.
56    pub outgoing_bytes_limit: u32,
57}
58
59impl ContextSettings {
60    /// Returns default settings with specified outgoing messages limits.
61    pub fn with_outgoing_limits(outgoing_limit: u32, outgoing_bytes_limit: u32) -> Self {
62        Self {
63            outgoing_limit,
64            outgoing_bytes_limit,
65            ..Default::default()
66        }
67    }
68}
69
70/// Dispatch or message with additional information.
71pub type OutgoingMessageInfo<T> = (T, u32, Option<ReservationId>);
72pub type OutgoingMessageInfoNoDelay<T> = (T, Option<ReservationId>);
73
74/// Context outcome dispatches and awakening ids.
75pub struct ContextOutcomeDrain {
76    /// Outgoing dispatches to be sent.
77    pub outgoing_dispatches: Vec<OutgoingMessageInfo<Dispatch>>,
78    /// Messages to be waken.
79    pub awakening: Vec<(MessageId, u32)>,
80    /// Reply deposits to be provided.
81    pub reply_deposits: Vec<(MessageId, u64)>,
82    /// Whether this execution sent out a reply.
83    pub reply_sent: bool,
84}
85
86/// Context outcome.
87///
88/// Contains all outgoing messages and wakes that should be done after execution.
89#[derive(Clone, Debug)]
90pub struct ContextOutcome {
91    init: Vec<OutgoingMessageInfo<InitMessage>>,
92    handle: Vec<OutgoingMessageInfo<HandleMessage>>,
93    reply: Option<OutgoingMessageInfoNoDelay<ReplyMessage>>,
94    // u32 is delay
95    awakening: Vec<(MessageId, u32)>,
96    // u64 is gas limit
97    // TODO: add Option<ReservationId> after #1828
98    reply_deposits: Vec<(MessageId, u64)>,
99    // Additional information section.
100    program_id: ActorId,
101    source: ActorId,
102    origin_msg_id: MessageId,
103}
104
105impl ContextOutcome {
106    /// Create new ContextOutcome.
107    fn new(program_id: ActorId, source: ActorId, origin_msg_id: MessageId) -> Self {
108        Self {
109            init: Vec::new(),
110            handle: Vec::new(),
111            reply: None,
112            awakening: Vec::new(),
113            reply_deposits: Vec::new(),
114            program_id,
115            source,
116            origin_msg_id,
117        }
118    }
119
120    /// Destructs outcome after execution and returns provided dispatches and awaken message ids.
121    pub fn drain(self) -> ContextOutcomeDrain {
122        let mut dispatches = Vec::new();
123        let reply_sent = self.reply.is_some();
124
125        for (msg, delay, reservation) in self.init.into_iter() {
126            dispatches.push((msg.into_dispatch(self.program_id), delay, reservation));
127        }
128
129        for (msg, delay, reservation) in self.handle.into_iter() {
130            dispatches.push((msg.into_dispatch(self.program_id), delay, reservation));
131        }
132
133        if let Some((msg, reservation)) = self.reply {
134            dispatches.push((
135                msg.into_dispatch(self.program_id, self.source, self.origin_msg_id),
136                0,
137                reservation,
138            ));
139        };
140
141        ContextOutcomeDrain {
142            outgoing_dispatches: dispatches,
143            awakening: self.awakening,
144            reply_deposits: self.reply_deposits,
145            reply_sent,
146        }
147    }
148}
149/// Store of current temporary message execution context.
150#[derive(Clone, Debug, Default, PartialEq, Eq, Hash, Decode, Encode, TypeInfo)]
151pub struct OutgoingPayloads {
152    handles: BTreeMap<u32, Option<Payload>>,
153    reply: Option<Payload>,
154    bytes_counter: u32,
155}
156
157/// Store of previous message execution context.
158#[derive(
159    Clone, Debug, Default, PartialEq, Eq, Hash, Decode, DecodeAsType, Encode, EncodeAsType, TypeInfo,
160)]
161#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
162pub struct ContextStore {
163    initialized: BTreeSet<ActorId>,
164    reservation_nonce: ReservationNonce,
165    system_reservation: Option<u64>,
166    /// Used to prevent creating messages with the same ID in DB. Before this was achieved by using `outgoing.len()`
167    /// but now it is moved to [OutgoingPayloads] thus we need to keep nonce here. Now to calculate nonce we simple increment `local_nonce`
168    /// in each `init` call.
169    local_nonce: u32,
170}
171
172impl ContextStore {
173    // TODO: Remove, only used in migrations (#issue 3721)
174    /// Create a new context store with the provided parameters.
175    pub fn new(
176        initialized: BTreeSet<ActorId>,
177        reservation_nonce: ReservationNonce,
178        system_reservation: Option<u64>,
179        local_nonce: u32,
180    ) -> Self {
181        Self {
182            initialized,
183            reservation_nonce,
184            system_reservation,
185            local_nonce,
186        }
187    }
188
189    /// Returns stored within message context reservation nonce.
190    ///
191    /// Will be non zero, if any reservations were created during
192    /// previous execution of the message.
193    pub(crate) fn reservation_nonce(&self) -> ReservationNonce {
194        self.reservation_nonce
195    }
196
197    /// Set reservation nonce from gas reserver.
198    ///
199    /// Gas reserver has actual nonce state during/after execution.
200    pub fn set_reservation_nonce(&mut self, gas_reserver: &GasReserver) {
201        self.reservation_nonce = gas_reserver.nonce();
202    }
203
204    /// Set system reservation.
205    pub fn add_system_reservation(&mut self, amount: u64) {
206        let reservation = &mut self.system_reservation;
207        *reservation = reservation
208            .map(|reservation| reservation.saturating_add(amount))
209            .or(Some(amount));
210    }
211
212    /// Get system reservation.
213    pub fn system_reservation(&self) -> Option<u64> {
214        self.system_reservation
215    }
216}
217
218/// Context of currently processing incoming message.
219#[derive(Clone, Debug)]
220pub struct MessageContext {
221    kind: DispatchKind,
222    current: IncomingMessage,
223    outcome: ContextOutcome,
224    store: ContextStore,
225    outgoing_payloads: OutgoingPayloads,
226    settings: ContextSettings,
227}
228
229impl MessageContext {
230    /// Create new message context.
231    /// Returns `None` if outgoing messages bytes limit exceeded.
232    pub fn new(dispatch: IncomingDispatch, program_id: ActorId, settings: ContextSettings) -> Self {
233        let (kind, message, store) = dispatch.into_parts();
234
235        Self {
236            kind,
237            outcome: ContextOutcome::new(program_id, message.source(), message.id()),
238            current: message,
239            store: store.unwrap_or_default(),
240            outgoing_payloads: OutgoingPayloads::default(),
241            settings,
242        }
243    }
244
245    /// Getter for inner settings.
246    pub fn settings(&self) -> &ContextSettings {
247        &self.settings
248    }
249
250    /// Getter for inner dispatch kind
251    pub fn kind(&self) -> DispatchKind {
252        self.kind
253    }
254
255    fn check_reply_availability(&self) -> Result<(), ExecutionError> {
256        if !matches!(self.kind, DispatchKind::Init | DispatchKind::Handle) {
257            return Err(ExecutionError::IncorrectEntryForReply);
258        }
259
260        Ok(())
261    }
262
263    fn increase_counter(counter: u32, amount: impl TryInto<u32>, limit: u32) -> Option<u32> {
264        TryInto::<u32>::try_into(amount)
265            .ok()
266            .and_then(|amount| counter.checked_add(amount))
267            .and_then(|counter| (counter <= limit).then_some(counter))
268    }
269
270    /// Return bool defining was reply sent within the execution.
271    pub fn reply_sent(&self) -> bool {
272        self.outcome.reply.is_some()
273    }
274
275    /// Send a new program initialization message.
276    ///
277    /// Generates a new message from provided data packet.
278    /// Returns message id and generated program id.
279    pub fn init_program(
280        &mut self,
281        packet: InitPacket,
282        delay: u32,
283    ) -> Result<(MessageId, ActorId), Error> {
284        let program_id = packet.destination();
285
286        if self.store.initialized.contains(&program_id) {
287            return Err(Error::DuplicateInit);
288        }
289
290        let last = self.store.local_nonce;
291
292        if last >= self.settings.outgoing_limit {
293            return Err(Error::OutgoingMessagesAmountLimitExceeded);
294        }
295
296        let message_id = MessageId::generate_outgoing(self.current.id(), last);
297        let message = InitMessage::from_packet(message_id, packet);
298        self.store.local_nonce += 1;
299        self.outgoing_payloads.handles.insert(last, None);
300        self.store.initialized.insert(program_id);
301        self.outcome.init.push((message, delay, None));
302
303        Ok((message_id, program_id))
304    }
305
306    /// Send a new program initialization message.
307    ///
308    /// Generates message from provided data packet and stored by handle payload.
309    /// Returns message id.
310    pub fn send_commit(
311        &mut self,
312        handle: u32,
313        mut packet: HandlePacket,
314        delay: u32,
315        reservation: Option<ReservationId>,
316    ) -> Result<MessageId, Error> {
317        let outgoing = self
318            .outgoing_payloads
319            .handles
320            .get_mut(&handle)
321            .ok_or(Error::OutOfBounds)?;
322        let data = outgoing.take().ok_or(Error::LateAccess)?;
323
324        let do_send_commit = || {
325            let Some(new_outgoing_bytes) = Self::increase_counter(
326                self.outgoing_payloads.bytes_counter,
327                packet.payload_len(),
328                self.settings.outgoing_bytes_limit,
329            ) else {
330                return Err((Error::OutgoingMessagesBytesLimitExceeded, data));
331            };
332
333            packet
334                .try_prepend(data)
335                .map_err(|data| (Error::MaxMessageSizeExceed, data))?;
336
337            let message_id = MessageId::generate_outgoing(self.current.id(), handle);
338            let message = HandleMessage::from_packet(message_id, packet);
339
340            self.outcome.handle.push((message, delay, reservation));
341
342            // Increasing `outgoing_bytes_counter`, instead of decreasing it,
343            // because this counter takes into account also messages,
344            // that are already committed during this execution.
345            // The message subsequent executions will recalculate this counter from
346            // store outgoing messages (see `Self::new`),
347            // so committed during this execution messages won't be taken into account
348            // during next executions.
349            self.outgoing_payloads.bytes_counter = new_outgoing_bytes;
350
351            Ok(message_id)
352        };
353
354        do_send_commit().map_err(|(err, data)| {
355            *outgoing = Some(data);
356            err
357        })
358    }
359
360    /// Provide space for storing payload for future message creation.
361    ///
362    /// Returns it's handle.
363    pub fn send_init(&mut self) -> Result<u32, Error> {
364        let last = self.store.local_nonce;
365        if last < self.settings.outgoing_limit {
366            self.store.local_nonce += 1;
367            self.outgoing_payloads
368                .handles
369                .insert(last, Some(Default::default()));
370
371            Ok(last)
372        } else {
373            Err(Error::OutgoingMessagesAmountLimitExceeded)
374        }
375    }
376
377    /// Pushes payload into stored payload by handle.
378    pub fn send_push(&mut self, handle: u32, buffer: &[u8]) -> Result<(), Error> {
379        let data = match self.outgoing_payloads.handles.get_mut(&handle) {
380            Some(Some(data)) => data,
381            Some(None) => return Err(Error::LateAccess),
382            None => return Err(Error::OutOfBounds),
383        };
384
385        let new_outgoing_bytes = Self::increase_counter(
386            self.outgoing_payloads.bytes_counter,
387            buffer.len(),
388            self.settings.outgoing_bytes_limit,
389        )
390        .ok_or(Error::OutgoingMessagesBytesLimitExceeded)?;
391
392        data.try_extend_from_slice(buffer)
393            .map_err(|_| Error::MaxMessageSizeExceed)?;
394
395        self.outgoing_payloads.bytes_counter = new_outgoing_bytes;
396
397        Ok(())
398    }
399
400    /// Pushes the incoming buffer/payload into stored payload by handle.
401    pub fn send_push_input(&mut self, handle: u32, range: CheckedRange) -> Result<(), Error> {
402        let data = match self.outgoing_payloads.handles.get_mut(&handle) {
403            Some(Some(data)) => data,
404            Some(None) => return Err(Error::LateAccess),
405            None => return Err(Error::OutOfBounds),
406        };
407
408        let bytes_amount = range.len();
409        let CheckedRange {
410            offset,
411            excluded_end,
412        } = range;
413
414        let new_outgoing_bytes = Self::increase_counter(
415            self.outgoing_payloads.bytes_counter,
416            bytes_amount,
417            self.settings.outgoing_bytes_limit,
418        )
419        .ok_or(Error::OutgoingMessagesBytesLimitExceeded)?;
420
421        data.try_extend_from_slice(&self.current.payload()[offset..excluded_end])
422            .map_err(|_| Error::MaxMessageSizeExceed)?;
423
424        self.outgoing_payloads.bytes_counter = new_outgoing_bytes;
425
426        Ok(())
427    }
428
429    /// Check if provided `offset`/`len` are correct for the current payload
430    /// limits. Result `CheckedRange` instance is accepted by
431    /// `send_push_input`/`reply_push_input` and has the method `len`
432    /// allowing to charge gas before the calls.
433    pub fn check_input_range(&self, offset: u32, len: u32) -> Result<CheckedRange, Error> {
434        let input_len = self.current.payload().len();
435        let offset = offset as usize;
436        let len = len as usize;
437
438        // Check `offset` is not out of bounds.
439        if offset >= input_len {
440            return Err(Error::OutOfBoundsInputSliceOffset);
441        }
442
443        // Check `len` for the current `offset` doesn't refer to the slice out of input bounds.
444        let available_len = input_len - offset;
445        if len > available_len {
446            return Err(Error::OutOfBoundsInputSliceLength);
447        }
448
449        Ok(CheckedRange {
450            offset,
451            // guaranteed to be `<= input.len()`, because of the check upper
452            excluded_end: offset.saturating_add(len),
453        })
454    }
455
456    /// Send reply message.
457    ///
458    /// Generates reply from provided data packet and stored reply payload.
459    /// Returns message id.
460    pub fn reply_commit(
461        &mut self,
462        mut packet: ReplyPacket,
463        reservation: Option<ReservationId>,
464    ) -> Result<MessageId, ExtError> {
465        self.check_reply_availability()?;
466
467        if self.reply_sent() {
468            return Err(Error::DuplicateReply.into());
469        }
470
471        let data = self.outgoing_payloads.reply.take().unwrap_or_default();
472
473        if let Err(data) = packet.try_prepend(data) {
474            self.outgoing_payloads.reply = Some(data);
475            return Err(Error::MaxMessageSizeExceed.into());
476        }
477
478        let message_id = MessageId::generate_reply(self.current.id());
479        let message = ReplyMessage::from_packet(message_id, packet);
480
481        self.outcome.reply = Some((message, reservation));
482
483        Ok(message_id)
484    }
485
486    /// Pushes payload into stored reply payload.
487    pub fn reply_push(&mut self, buffer: &[u8]) -> Result<(), ExtError> {
488        self.check_reply_availability()?;
489
490        if self.reply_sent() {
491            return Err(Error::LateAccess.into());
492        }
493
494        // NOTE: it's normal to not undone `get_or_insert_with` in case of error
495        self.outgoing_payloads
496            .reply
497            .get_or_insert_with(Default::default)
498            .try_extend_from_slice(buffer)
499            .map_err(|_| Error::MaxMessageSizeExceed.into())
500    }
501
502    /// Return reply destination.
503    pub fn reply_destination(&self) -> ActorId {
504        self.outcome.source
505    }
506
507    /// Pushes the incoming message buffer into stored reply payload.
508    pub fn reply_push_input(&mut self, range: CheckedRange) -> Result<(), ExtError> {
509        self.check_reply_availability()?;
510
511        if self.reply_sent() {
512            return Err(Error::LateAccess.into());
513        }
514
515        let CheckedRange {
516            offset,
517            excluded_end,
518        } = range;
519
520        // NOTE: it's normal to not undone `get_or_insert_with` in case of error
521        self.outgoing_payloads
522            .reply
523            .get_or_insert_with(Default::default)
524            .try_extend_from_slice(&self.current.payload()[offset..excluded_end])
525            .map_err(|_| Error::MaxMessageSizeExceed.into())
526    }
527
528    /// Wake message by it's message id.
529    pub fn wake(&mut self, waker_id: MessageId, delay: u32) -> Result<(), Error> {
530        if !self.outcome.awakening.iter().any(|v| v.0 == waker_id) {
531            self.outcome.awakening.push((waker_id, delay));
532            Ok(())
533        } else {
534            Err(Error::DuplicateWaking)
535        }
536    }
537
538    /// Create deposit to handle future reply on message id was sent.
539    pub fn reply_deposit(
540        &mut self,
541        message_id: MessageId,
542        amount: u64,
543    ) -> Result<(), MessageError> {
544        if self
545            .outcome
546            .reply_deposits
547            .iter()
548            .any(|(mid, _)| mid == &message_id)
549        {
550            return Err(MessageError::DuplicateReplyDeposit);
551        }
552
553        if !self
554            .outcome
555            .handle
556            .iter()
557            .any(|(message, ..)| message.id() == message_id)
558            && !self
559                .outcome
560                .init
561                .iter()
562                .any(|(message, ..)| message.id() == message_id)
563        {
564            return Err(MessageError::IncorrectMessageForReplyDeposit);
565        }
566
567        self.outcome.reply_deposits.push((message_id, amount));
568
569        Ok(())
570    }
571
572    /// Current processing incoming message.
573    pub fn current(&self) -> &IncomingMessage {
574        &self.current
575    }
576
577    /// Current program's id.
578    pub fn program_id(&self) -> ActorId {
579        self.outcome.program_id
580    }
581
582    /// Destructs context after execution and returns provided outcome and store.
583    pub fn drain(self) -> (ContextOutcome, ContextStore) {
584        let Self { outcome, store, .. } = self;
585
586        (outcome, store)
587    }
588}
589
590pub struct CheckedRange {
591    offset: usize,
592    excluded_end: usize,
593}
594
595impl CheckedRange {
596    pub fn len(&self) -> u32 {
597        (self.excluded_end - self.offset) as u32
598    }
599}
600
601#[cfg(test)]
602mod tests {
603    use super::*;
604    use alloc::vec;
605    use core::convert::TryInto;
606
607    macro_rules! assert_ok {
608        ( $x:expr $(,)? ) => {
609            let is = $x;
610            match is {
611                Ok(_) => (),
612                _ => assert!(false, "Expected Ok(_). Got {:#?}", is),
613            }
614        };
615        ( $x:expr, $y:expr $(,)? ) => {
616            assert_eq!($x, Ok($y));
617        };
618    }
619
620    macro_rules! assert_err {
621        ( $x:expr , $y:expr $(,)? ) => {
622            assert_eq!($x, Err($y.into()));
623        };
624    }
625
626    // Set of constants for clarity of a part of the test
627    const INCOMING_MESSAGE_ID: u64 = 3;
628    const INCOMING_MESSAGE_SOURCE: u64 = 4;
629
630    #[test]
631    fn duplicated_init() {
632        let mut message_context = MessageContext::new(
633            Default::default(),
634            Default::default(),
635            ContextSettings::with_outgoing_limits(1024, u32::MAX),
636        );
637
638        // first init to default ActorId.
639        assert_ok!(message_context.init_program(Default::default(), 0));
640
641        // second init to same default ActorId should get error.
642        assert_err!(
643            message_context.init_program(Default::default(), 0),
644            Error::DuplicateInit,
645        );
646    }
647
648    #[test]
649    fn send_push_bytes_exceeded() {
650        let mut message_context = MessageContext::new(
651            Default::default(),
652            Default::default(),
653            ContextSettings::with_outgoing_limits(1024, 10),
654        );
655
656        let handle = message_context.send_init().unwrap();
657
658        // push 5 bytes
659        assert_ok!(message_context.send_push(handle, &[1, 2, 3, 4, 5]));
660
661        // push 5 bytes
662        assert_ok!(message_context.send_push(handle, &[1, 2, 3, 4, 5]));
663
664        // push 1 byte should get error.
665        assert_err!(
666            message_context.send_push(handle, &[1]),
667            Error::OutgoingMessagesBytesLimitExceeded,
668        );
669    }
670
671    #[test]
672    fn send_commit_bytes_exceeded() {
673        let mut message_context = MessageContext::new(
674            Default::default(),
675            Default::default(),
676            ContextSettings::with_outgoing_limits(1024, 10),
677        );
678
679        let handle = message_context.send_init().unwrap();
680
681        // push 5 bytes
682        assert_ok!(message_context.send_push(handle, &[1, 2, 3, 4, 5]));
683
684        // commit 6 bytes should get error.
685        assert_err!(
686            message_context.send_commit(
687                handle,
688                HandlePacket::new(
689                    Default::default(),
690                    Payload::try_from([1, 2, 3, 4, 5, 6].to_vec()).unwrap(),
691                    0
692                ),
693                0,
694                None
695            ),
696            Error::OutgoingMessagesBytesLimitExceeded,
697        );
698
699        // commit 5 bytes should be ok.
700        assert_ok!(message_context.send_commit(
701            handle,
702            HandlePacket::new(
703                Default::default(),
704                Payload::try_from([1, 2, 3, 4, 5].to_vec()).unwrap(),
705                0,
706            ),
707            0,
708            None,
709        ));
710
711        let messages = message_context.drain().0.drain().outgoing_dispatches;
712        assert_eq!(
713            messages[0].0.payload_bytes(),
714            [1, 2, 3, 4, 5, 1, 2, 3, 4, 5]
715        );
716    }
717
718    #[test]
719    fn send_commit_message_size_limit() {
720        let mut message_context = MessageContext::new(
721            Default::default(),
722            Default::default(),
723            ContextSettings::with_outgoing_limits(1024, u32::MAX),
724        );
725
726        let handle = message_context.send_init().unwrap();
727
728        // push 1 byte
729        assert_ok!(message_context.send_push(handle, &[1]));
730
731        let payload = Payload::repeat(2);
732        assert_err!(
733            message_context.send_commit(
734                handle,
735                HandlePacket::new(Default::default(), payload, 0),
736                0,
737                None
738            ),
739            Error::MaxMessageSizeExceed,
740        );
741
742        let payload = Payload::try_from(vec![1; Payload::MAX_LEN - 1]).unwrap();
743        assert_ok!(message_context.send_commit(
744            handle,
745            HandlePacket::new(Default::default(), payload, 0),
746            0,
747            None,
748        ));
749
750        let messages = message_context.drain().0.drain().outgoing_dispatches;
751        assert_eq!(
752            Payload::try_from(messages[0].0.payload_bytes().to_vec()).unwrap(),
753            Payload::repeat(1)
754        );
755    }
756
757    #[test]
758    fn send_push_input_bytes_exceeded() {
759        let incoming_message = IncomingMessage::new(
760            MessageId::from(INCOMING_MESSAGE_ID),
761            ActorId::from(INCOMING_MESSAGE_SOURCE),
762            vec![1, 2, 3, 4, 5].try_into().unwrap(),
763            0,
764            0,
765            None,
766        );
767
768        let incoming_dispatch = IncomingDispatch::new(DispatchKind::Handle, incoming_message, None);
769
770        // Creating a message context
771        let mut message_context = MessageContext::new(
772            incoming_dispatch,
773            Default::default(),
774            ContextSettings::with_outgoing_limits(1024, 10),
775        );
776
777        let handle = message_context.send_init().unwrap();
778
779        // push 5 bytes
780        assert_ok!(message_context.send_push_input(
781            handle,
782            CheckedRange {
783                offset: 0,
784                excluded_end: 5,
785            }
786        ));
787
788        // push 5 bytes
789        assert_ok!(message_context.send_push_input(
790            handle,
791            CheckedRange {
792                offset: 0,
793                excluded_end: 5,
794            }
795        ));
796
797        // push 1 byte should get error.
798        assert_err!(
799            message_context.send_push_input(
800                handle,
801                CheckedRange {
802                    offset: 0,
803                    excluded_end: 1,
804                }
805            ),
806            Error::OutgoingMessagesBytesLimitExceeded,
807        );
808    }
809
810    #[test]
811    fn outgoing_limit_exceeded() {
812        // Check that we can always send exactly outgoing_limit messages.
813        let max_n = 5;
814
815        for n in 0..=max_n {
816            // for outgoing_limit n checking that LimitExceeded will be after n's message.
817            let settings = ContextSettings::with_outgoing_limits(n, u32::MAX);
818
819            let mut message_context =
820                MessageContext::new(Default::default(), Default::default(), settings);
821            // send n messages
822            for _ in 0..n {
823                let handle = message_context.send_init().expect("unreachable");
824                message_context
825                    .send_push(handle, b"payload")
826                    .expect("unreachable");
827                message_context
828                    .send_commit(handle, HandlePacket::default(), 0, None)
829                    .expect("unreachable");
830            }
831            // n + 1 should get first error.
832            let limit_exceeded = message_context.send_init();
833            assert_eq!(
834                limit_exceeded,
835                Err(Error::OutgoingMessagesAmountLimitExceeded)
836            );
837
838            // we can't send messages in this MessageContext.
839            let limit_exceeded = message_context.init_program(Default::default(), 0);
840            assert_eq!(
841                limit_exceeded,
842                Err(Error::OutgoingMessagesAmountLimitExceeded)
843            );
844        }
845    }
846
847    #[test]
848    fn invalid_out_of_bounds() {
849        let mut message_context = MessageContext::new(
850            Default::default(),
851            Default::default(),
852            ContextSettings::with_outgoing_limits(1024, u32::MAX),
853        );
854
855        // Use invalid handle 0.
856        let out_of_bounds = message_context.send_commit(0, Default::default(), 0, None);
857        assert_eq!(out_of_bounds, Err(Error::OutOfBounds));
858
859        // make 0 valid.
860        let valid_handle = message_context.send_init().expect("unreachable");
861        assert_eq!(valid_handle, 0);
862
863        // Use valid handle 0.
864        assert_ok!(message_context.send_commit(0, Default::default(), 0, None));
865
866        // Use invalid handle 42.
867        assert_err!(
868            message_context.send_commit(42, Default::default(), 0, None),
869            Error::OutOfBounds,
870        );
871    }
872
873    #[test]
874    fn double_reply() {
875        let mut message_context = MessageContext::new(
876            Default::default(),
877            Default::default(),
878            ContextSettings::with_outgoing_limits(1024, u32::MAX),
879        );
880
881        // First reply.
882        assert_ok!(message_context.reply_commit(Default::default(), None));
883
884        // Reply twice in one message is forbidden.
885        assert_err!(
886            message_context.reply_commit(Default::default(), None),
887            Error::DuplicateReply,
888        );
889    }
890
891    #[test]
892    fn reply_commit_message_size_limit() {
893        let mut message_context =
894            MessageContext::new(Default::default(), Default::default(), Default::default());
895
896        assert_ok!(message_context.reply_push(&[1]));
897
898        let payload = Payload::repeat(2);
899        assert_err!(
900            message_context.reply_commit(ReplyPacket::new(payload, 0), None),
901            Error::MaxMessageSizeExceed,
902        );
903
904        let payload = Payload::try_from(vec![1; Payload::MAX_LEN - 1]).unwrap();
905        assert_ok!(message_context.reply_commit(ReplyPacket::new(payload, 0), None));
906
907        let messages = message_context.drain().0.drain().outgoing_dispatches;
908        assert_eq!(
909            Payload::try_from(messages[0].0.payload_bytes().to_vec()).unwrap(),
910            Payload::repeat(1)
911        );
912    }
913
914    #[test]
915    /// Test that covers full api of `MessageContext`
916    fn message_context_api() {
917        // Creating an incoming message around which the runner builds the `MessageContext`
918        let incoming_message = IncomingMessage::new(
919            MessageId::from(INCOMING_MESSAGE_ID),
920            ActorId::from(INCOMING_MESSAGE_SOURCE),
921            vec![1, 2].try_into().unwrap(),
922            0,
923            0,
924            None,
925        );
926
927        let incoming_dispatch = IncomingDispatch::new(DispatchKind::Handle, incoming_message, None);
928
929        // Creating a message context
930        let mut context = MessageContext::new(
931            incoming_dispatch,
932            Default::default(),
933            ContextSettings::with_outgoing_limits(1024, u32::MAX),
934        );
935
936        // Checking that the initial parameters of the context match the passed constants
937        assert_eq!(context.current().id(), MessageId::from(INCOMING_MESSAGE_ID));
938
939        // Creating a reply packet
940        let reply_packet = ReplyPacket::new(vec![0, 0].try_into().unwrap(), 0);
941
942        // Checking that we are able to initialize reply
943        assert_ok!(context.reply_push(&[1, 2, 3]));
944
945        // Setting reply message and making sure the operation was successful
946        assert_ok!(context.reply_commit(reply_packet.clone(), None));
947
948        // Checking that the `ReplyMessage` matches the passed one
949        assert_eq!(
950            context
951                .outcome
952                .reply
953                .as_ref()
954                .unwrap()
955                .0
956                .payload_bytes()
957                .to_vec(),
958            vec![1, 2, 3, 0, 0],
959        );
960
961        // Checking that repeated call `reply_push(...)` returns error and does not do anything
962        assert_err!(context.reply_push(&[1]), Error::LateAccess);
963        assert_eq!(
964            context
965                .outcome
966                .reply
967                .as_ref()
968                .unwrap()
969                .0
970                .payload_bytes()
971                .to_vec(),
972            vec![1, 2, 3, 0, 0],
973        );
974
975        // Checking that repeated call `reply_commit(...)` returns error and does not
976        assert_err!(
977            context.reply_commit(reply_packet, None),
978            Error::DuplicateReply
979        );
980
981        // Checking that at this point vector of outgoing messages is empty
982        assert!(context.outcome.handle.is_empty());
983
984        // Creating an expected handle for a future initialized message
985        let expected_handle = 0;
986
987        // Initializing message and compare its handle with expected one
988        assert_eq!(
989            context.send_init().expect("Error initializing new message"),
990            expected_handle
991        );
992
993        // And checking that it is not formed
994        assert!(
995            context
996                .outgoing_payloads
997                .handles
998                .get(&expected_handle)
999                .expect("This key should be")
1000                .is_some()
1001        );
1002
1003        // Checking that we are able to push payload for the
1004        // message that we have not committed yet
1005        assert_ok!(context.send_push(expected_handle, &[5, 7]));
1006        assert_ok!(context.send_push(expected_handle, &[9]));
1007
1008        // Creating an outgoing packet to commit sending by parts
1009        let commit_packet = HandlePacket::default();
1010
1011        // Checking if commit is successful
1012        assert_ok!(context.send_commit(expected_handle, commit_packet, 0, None));
1013
1014        // Checking that we are **NOT** able to push payload for the message or
1015        // commit it if we already committed it or directly pushed before
1016        assert_err!(
1017            context.send_push(expected_handle, &[5, 7]),
1018            Error::LateAccess,
1019        );
1020        assert_err!(
1021            context.send_commit(expected_handle, HandlePacket::default(), 0, None),
1022            Error::LateAccess,
1023        );
1024
1025        // Creating a handle to push and do commit non-existent message
1026        let expected_handle = 15;
1027
1028        // Checking that we also get an error when trying
1029        // to commit or send a non-existent message
1030        assert_err!(context.send_push(expected_handle, &[0]), Error::OutOfBounds);
1031        assert_err!(
1032            context.send_commit(expected_handle, HandlePacket::default(), 0, None),
1033            Error::OutOfBounds,
1034        );
1035
1036        // Creating a handle to init and do not commit later
1037        // to show that the message will not be sent
1038        let expected_handle = 1;
1039
1040        assert_eq!(
1041            context.send_init().expect("Error initializing new message"),
1042            expected_handle
1043        );
1044        assert_ok!(context.send_push(expected_handle, &[2, 2]));
1045
1046        // Checking that reply message not lost and matches our initial
1047        assert!(context.outcome.reply.is_some());
1048        assert_eq!(
1049            context.outcome.reply.as_ref().unwrap().0.payload_bytes(),
1050            vec![1, 2, 3, 0, 0]
1051        );
1052
1053        // Checking that on drain we get only messages that were fully formed (directly sent or committed)
1054        let (expected_result, _) = context.drain();
1055        assert_eq!(expected_result.handle.len(), 1);
1056        assert_eq!(expected_result.handle[0].0.payload_bytes(), vec![5, 7, 9]);
1057    }
1058
1059    #[test]
1060    fn duplicate_waking() {
1061        let incoming_message = IncomingMessage::new(
1062            MessageId::from(INCOMING_MESSAGE_ID),
1063            ActorId::from(INCOMING_MESSAGE_SOURCE),
1064            vec![1, 2].try_into().unwrap(),
1065            0,
1066            0,
1067            None,
1068        );
1069
1070        let incoming_dispatch = IncomingDispatch::new(DispatchKind::Handle, incoming_message, None);
1071
1072        let mut context = MessageContext::new(
1073            incoming_dispatch,
1074            Default::default(),
1075            ContextSettings::with_outgoing_limits(1024, u32::MAX),
1076        );
1077
1078        context.wake(MessageId::default(), 10).unwrap();
1079
1080        assert_eq!(
1081            context.wake(MessageId::default(), 1),
1082            Err(Error::DuplicateWaking)
1083        );
1084    }
1085
1086    #[test]
1087    fn duplicate_reply_deposit() {
1088        let incoming_message = IncomingMessage::new(
1089            MessageId::from(INCOMING_MESSAGE_ID),
1090            ActorId::from(INCOMING_MESSAGE_SOURCE),
1091            vec![1, 2].try_into().unwrap(),
1092            0,
1093            0,
1094            None,
1095        );
1096
1097        let incoming_dispatch = IncomingDispatch::new(DispatchKind::Handle, incoming_message, None);
1098
1099        let mut message_context = MessageContext::new(
1100            incoming_dispatch,
1101            Default::default(),
1102            ContextSettings::with_outgoing_limits(1024, u32::MAX),
1103        );
1104
1105        let handle = message_context.send_init().expect("unreachable");
1106        message_context
1107            .send_push(handle, b"payload")
1108            .expect("unreachable");
1109        let message_id = message_context
1110            .send_commit(handle, HandlePacket::default(), 0, None)
1111            .expect("unreachable");
1112
1113        assert!(message_context.reply_deposit(message_id, 1234).is_ok());
1114        assert_err!(
1115            message_context.reply_deposit(message_id, 1234),
1116            MessageError::DuplicateReplyDeposit
1117        );
1118    }
1119
1120    #[test]
1121    fn inexistent_reply_deposit() {
1122        let incoming_message = IncomingMessage::new(
1123            MessageId::from(INCOMING_MESSAGE_ID),
1124            ActorId::from(INCOMING_MESSAGE_SOURCE),
1125            vec![1, 2].try_into().unwrap(),
1126            0,
1127            0,
1128            None,
1129        );
1130
1131        let incoming_dispatch = IncomingDispatch::new(DispatchKind::Handle, incoming_message, None);
1132
1133        let mut message_context = MessageContext::new(
1134            incoming_dispatch,
1135            Default::default(),
1136            ContextSettings::with_outgoing_limits(1024, u32::MAX),
1137        );
1138
1139        let message_id = message_context
1140            .reply_commit(ReplyPacket::default(), None)
1141            .expect("unreachable");
1142
1143        assert_err!(
1144            message_context.reply_deposit(message_id, 1234),
1145            MessageError::IncorrectMessageForReplyDeposit
1146        );
1147        assert_err!(
1148            message_context.reply_deposit(Default::default(), 1234),
1149            MessageError::IncorrectMessageForReplyDeposit
1150        );
1151    }
1152}
gear_core/message/context.rs

gear_core/message/
context.rs
