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

gear_core/message/
context.rs
