1use super::*;
3use std::cell::Cell;
4use std::sync::{
5 Weak,
6 atomic::{AtomicU8, AtomicU64, Ordering},
7};
8static SERIAL: AtomicU64 = AtomicU64::new(1);
9pub(super) fn serial() -> u64 {
10 next_serial(&SERIAL)
11}
12fn next_serial(counter: &AtomicU64) -> u64 {
13 let mut n = counter.load(Ordering::SeqCst);
14 loop {
15 let Some(next) = n.checked_add(1) else {
16 return 0;
17 };
18 match counter.compare_exchange(n, next, Ordering::SeqCst, Ordering::SeqCst) {
19 Ok(_) => return n,
20 Err(current) => n = current,
21 }
22 }
23}
24#[derive(Clone, Copy)]
25struct Intent {
26 router: u64,
27 parent: u64,
28 birth: u64,
29 retry: bool,
30}
31thread_local! { static INTENT: Cell<Option<Intent>> = const { Cell::new(None) }; }
32pub(super) fn is_retry_intent() -> bool {
33 INTENT.get().is_some_and(|i| i.retry)
34}
35pub(super) struct IntentGuard(Option<Intent>);
36impl Drop for IntentGuard {
37 fn drop(&mut self) {
38 INTENT.set(self.0);
39 }
40}
41#[derive(Clone, Copy, Debug)]
43pub struct RetryTrackerLimits {
44 pub max_pending: usize,
45 pub max_pending_bytes: usize,
46 pub history_entries: usize,
47 pub history_bytes: usize,
48}
49impl Default for RetryTrackerLimits {
50 fn default() -> Self {
51 Self {
52 max_pending: 32,
53 max_pending_bytes: 2048,
54 history_entries: 4,
55 history_bytes: 768,
56 }
57 }
58}
59struct History {
60 parent: u64,
61 birth: u64,
62 bytes: Arc<[u8]>,
63}
64struct Ledger {
65 fence: u64,
66 pending: Vec<Weak<Write>>,
67 history: Vec<History>,
68}
69pub(super) struct TrackerInner {
70 router: u64,
71 side: AtomicU64,
72 limits: RetryTrackerLimits,
73 ledger: RouterMutex<Ledger>,
74}
75#[derive(Clone)]
77pub struct RetryTracker(Arc<TrackerInner>);
78impl PartialEq for RetryTracker {
79 fn eq(&self, other: &Self) -> bool {
80 Arc::ptr_eq(&self.0, &other.0)
81 }
82}
83impl Eq for RetryTracker {}
84impl Debug for RetryTracker {
85 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
86 f.debug_struct("RetryTracker").finish_non_exhaustive()
87 }
88}
89const ADMITTED: u8 = 1;
90const STARTED: u8 = 2;
91const DRAINED: u8 = 4;
92const RETIRED: u8 = 8;
93const DISCARDED: u8 = 16;
94struct Write {
95 bytes: Arc<[u8]>,
96 fence: u64,
97 parent: u64,
98 birth: u64,
99 eligible: bool,
100 admitted: AtomicU64,
101 flags: AtomicU8,
102}
103#[derive(Clone)]
106pub struct PackedWriteLease {
107 write: Arc<Write>,
108 tracker: Option<Arc<TrackerInner>>,
109}
110impl Debug for PackedWriteLease {
111 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
112 f.debug_struct("PackedWriteLease")
113 .field("bytes", &self.write.bytes.len())
114 .finish_non_exhaustive()
115 }
116}
117#[derive(Clone, Copy, Debug, PartialEq, Eq)]
119pub enum WriteStart {
120 Ready,
121 PendingAdmission,
122 Retired,
123}
124impl RetryTracker {
125 pub fn invalidate_generation(&self) -> TelemetryResult<()> {
126 let mut l = self.0.ledger.lock();
127 l.fence = serial();
128 l.history.clear();
129 Ok(())
130 }
131 fn offer(&self, bytes: &[u8]) -> PackedWriteLease {
132 let owned_bytes: Arc<[u8]> = Arc::from(bytes);
134 let mut l = self.0.ledger.lock();
135 l.pending.retain(|w| w.strong_count() != 0);
136 let intent = INTENT.get().filter(|i| {
137 i.router == self.0.router && i.birth != 0 && l.fence != 0 && i.birth > l.fence
138 });
139 let used = l
140 .pending
141 .iter()
142 .filter_map(Weak::upgrade)
143 .map(|w| w.bytes.len())
144 .sum::<usize>();
145 let tracked = intent.filter(|_| {
146 l.pending.len() < self.0.limits.max_pending
147 && used
148 .checked_add(bytes.len())
149 .is_some_and(|n| n <= self.0.limits.max_pending_bytes)
150 && canonical(bytes)
151 });
152 let eligible = tracked.is_some_and(|i| {
153 i.retry
154 && l.history.iter().any(|h| {
155 h.parent == i.parent && h.birth == i.birth && h.bytes.as_ref() == bytes
156 })
157 });
158 let w = Arc::new(Write {
159 bytes: owned_bytes,
160 fence: l.fence,
161 parent: tracked.map_or(0, |i| i.parent),
162 birth: tracked.map_or(0, |i| i.birth),
163 eligible,
164 admitted: AtomicU64::new(0),
165 flags: AtomicU8::new(0),
166 });
167 if tracked.is_some() {
168 l.pending.push(Arc::downgrade(&w));
169 }
170 PackedWriteLease {
171 write: w,
172 tracker: Some(self.0.clone()),
173 }
174 }
175}
176impl PackedWriteLease {
177 pub fn bytes(&self) -> &[u8] {
178 &self.write.bytes
179 }
180 pub fn admit(&self) -> TelemetryResult<()> {
182 if let Some(t) = &self.tracker {
183 let _l = t.ledger.lock();
184 if self.write.flags.load(Ordering::Acquire) & ADMITTED == 0 {
185 self.write.admitted.store(serial(), Ordering::Release);
186 self.write.flags.fetch_or(ADMITTED, Ordering::AcqRel);
187 }
188 } else {
189 self.write.flags.fetch_or(ADMITTED, Ordering::AcqRel);
190 }
191 Ok(())
192 }
193 pub fn discard_if_retired(&self) -> bool {
194 let Some(t) = &self.tracker else { return false };
195 let l = t.ledger.lock();
196 let f = self.write.flags.load(Ordering::Acquire);
197 if f & DISCARDED != 0 {
198 return true;
199 }
200 if l.fence != 0 && l.fence == self.write.fence && f & RETIRED != 0 && f & STARTED == 0 {
201 self.write.flags.fetch_or(DISCARDED, Ordering::AcqRel);
202 true
203 } else {
204 false
205 }
206 }
207 pub fn try_start(&self) -> WriteStart {
208 let _l = self.tracker.as_ref().map(|t| t.ledger.lock());
209 let f = self.write.flags.load(Ordering::Acquire);
210 if f & DISCARDED != 0
211 || (f & RETIRED != 0
212 && f & STARTED == 0
213 && _l
214 .as_ref()
215 .is_some_and(|l| l.fence != 0 && l.fence == self.write.fence))
216 {
217 self.write.flags.fetch_or(DISCARDED, Ordering::AcqRel);
218 return WriteStart::Retired;
219 }
220 if f & ADMITTED == 0 {
221 return WriteStart::PendingAdmission;
222 }
223 self.write.flags.fetch_or(STARTED, Ordering::AcqRel);
224 WriteStart::Ready
225 }
226 pub fn try_start_batch(leases: &[&Self]) -> TelemetryResult<WriteStart> {
229 let Some(first) = leases.first() else {
230 return Ok(WriteStart::Ready);
231 };
232 for (index, lease) in leases.iter().enumerate() {
233 if leases[..index]
234 .iter()
235 .any(|other| Arc::ptr_eq(&other.write, &lease.write))
236 {
237 return Err(TelemetryError::BadArg);
238 }
239 match (&first.tracker, &lease.tracker) {
240 (Some(a), Some(b)) if Arc::ptr_eq(a, b) => {}
241 (None, None) => {}
242 _ => return Err(TelemetryError::BadArg),
243 }
244 }
245 let ledger = first.tracker.as_ref().map(|t| t.ledger.lock());
246 for lease in leases {
247 let flags = lease.write.flags.load(Ordering::Acquire);
248 if flags & DISCARDED != 0
249 || (flags & RETIRED != 0
250 && flags & STARTED == 0
251 && ledger
252 .as_ref()
253 .is_some_and(|l| l.fence != 0 && l.fence == lease.write.fence))
254 {
255 return Ok(WriteStart::Retired);
256 }
257 }
258 if leases
259 .iter()
260 .any(|lease| lease.write.flags.load(Ordering::Acquire) & ADMITTED == 0)
261 {
262 return Ok(WriteStart::PendingAdmission);
263 }
264 for lease in leases {
265 lease.write.flags.fetch_or(STARTED, Ordering::AcqRel);
266 }
267 Ok(WriteStart::Ready)
268 }
269 pub fn report_drained(&self) -> TelemetryResult<()> {
271 let Some(t) = &self.tracker else {
272 return Ok(());
273 };
274 let mut l = t.ledger.lock();
275 let f = self.write.flags.load(Ordering::Acquire);
276 if f & STARTED == 0 || f & DISCARDED != 0 {
277 return Err(TelemetryError::BadArg);
278 }
279 if f & DRAINED != 0 {
280 return Ok(());
281 }
282 self.write.flags.fetch_or(DRAINED, Ordering::AcqRel);
283 if self.write.birth == 0
284 || l.fence == 0
285 || l.fence != self.write.fence
286 || self.write.bytes.len() > t.limits.history_bytes
287 || t.limits.history_entries == 0
288 {
289 return Ok(());
290 }
291 while !l.history.is_empty()
292 && (l.history.len() >= t.limits.history_entries
293 || l.history
294 .iter()
295 .map(|h| h.bytes.len())
296 .sum::<usize>()
297 .saturating_add(self.write.bytes.len())
298 > t.limits.history_bytes)
299 {
300 l.history.remove(0);
301 }
302 l.history.push(History {
303 parent: self.write.parent,
304 birth: self.write.birth,
305 bytes: self.write.bytes.clone(),
306 });
307 Ok(())
308 }
309}
310fn canonical(bytes: &[u8]) -> bool {
311 bytes.first().is_some_and(|flags| flags & 0x11 == 0)
312 && wire_format::peek_routing_frame_info(bytes).is_ok_and(|f| {
313 f.reliable.is_some_and(|h| {
314 h.flags == wire_format::RELIABLE_FLAG_UNSEQUENCED && h.seq == 0 && h.ack == 0
315 }) && f.envelope.wire_shape.is_some()
316 && f.envelope.target_senders().len() != 0
317 })
318}
319fn same_contract(id: u64, data: &RouterItem, bytes: &[u8]) -> bool {
320 let Ok(retry) = wire_format::unpack_packet(bytes) else {
321 return false;
322 };
323 let same = |original: &Packet| {
324 original.packet_id() == id
325 && retry.packet_id() == id
326 && original.data_type() == retry.data_type()
327 && crate::packet::sender_address_u32(original.sender())
328 == crate::packet::sender_address_u32(retry.sender())
329 && original.timestamp() == retry.timestamp()
330 && original.nonce() == retry.nonce()
331 && original.data_size() == retry.data_size()
332 && original.payload() == retry.payload()
333 && original.wire_shape().is_some()
334 && original.wire_shape() == retry.wire_shape()
335 && !original.wire_target_senders().is_empty()
336 && original.wire_target_senders() == retry.wire_target_senders()
337 && wire_format::unique_sorted_endpoints(original.endpoints())
338 .eq(wire_format::unique_sorted_endpoints(retry.endpoints()))
339 };
340 match data {
341 RouterItem::Packet(original) => same(original),
342 RouterItem::Packed(original) => {
343 wire_format::unpack_packet(original.as_ref()).is_ok_and(|packet| same(&packet))
344 }
345 }
346}
347impl Router {
348 pub fn add_side_packed_leased_with_priority_and_options<N, F>(
352 &self,
353 name: N,
354 tx: F,
355 options: RouterSideOptions,
356 limits: RetryTrackerLimits,
357 ) -> TelemetryResult<(RouterSideId, RetryTracker)>
358 where
359 N: AsRef<str>,
360 F: Fn(PackedWriteLease, u8) -> TelemetryResult<()> + Send + Sync + 'static,
361 {
362 let mut registry = self.retry_trackers.lock();
363 registry.retain(|t| t.strong_count() != 0);
364 let router = registry
365 .iter()
366 .find_map(Weak::upgrade)
367 .map_or_else(serial, |t| t.router);
368 let fence = serial();
369 let tracker = RetryTracker(Arc::new(TrackerInner {
370 router,
371 side: AtomicU64::new(u64::MAX),
372 limits,
373 ledger: RouterMutex::new(Ledger {
374 fence,
375 pending: Vec::new(),
376 history: Vec::new(),
377 }),
378 }));
379 registry.push(Arc::downgrade(&tracker.0));
380 drop(registry);
381 let callback_tracker = tracker.clone();
382 let side = self.add_side_packed_with_priority_and_options(
383 name,
384 move |bytes, priority| tx(callback_tracker.offer(bytes), priority),
385 options,
386 );
387 tracker.0.side.store(side as u64, Ordering::Release);
388 tracker.invalidate_generation()?;
391 Ok((side, tracker))
392 }
393 #[cfg(feature = "discovery")]
394 pub(super) fn retry_birth(&self) -> u64 {
395 if self
396 .retry_trackers
397 .lock()
398 .iter()
399 .any(|w| w.strong_count() != 0)
400 {
401 serial()
402 } else {
403 0
404 }
405 }
406 pub(super) fn retry_queued_intent(&self, claim: Option<TxClaim>) -> IntentGuard {
407 let queued = claim.filter(|c| c.side_retry).and_then(|c| {
408 let st = self.state.lock();
409 let index = st.claim_index(c.token)?;
410 let q = st.transmit_queue.get(index)?;
411 Some((q.deferred_since_ms, q.disposable & TX_E2E_RETRY != 0))
412 });
413 self.retry_intent(queued.map(|q| q.0), queued.is_some_and(|q| q.1))
414 }
415 pub(super) fn retry_intent(&self, parent: Option<u64>, retry: bool) -> IntentGuard {
416 let next = parent.and_then(|parent| {
417 let birth = self
418 .state
419 .lock()
420 .end_to_end_reliable_tx
421 .get(&parent)?
422 .completion_epoch;
423 let router = self
424 .retry_trackers
425 .lock()
426 .iter()
427 .find_map(Weak::upgrade)?
428 .router;
429 Some(Intent {
430 router,
431 parent,
432 birth,
433 retry,
434 })
435 });
436 IntentGuard(INTENT.replace(next))
437 }
438 pub(super) fn invalidate_retry_side(&self, side: RouterSideId) {
439 self.retry_trackers.lock().retain(|weak| {
440 let Some(t) = weak.upgrade() else {
441 return false;
442 };
443 if t.side.load(Ordering::Acquire) == side as u64 {
444 let _ = RetryTracker(t).invalidate_generation();
445 false
446 } else {
447 true
448 }
449 });
450 }
451 pub(super) fn retire_completed_writes(
452 &self,
453 id: u64,
454 completed: &EndToEndReliableSent,
455 cutoff: u64,
456 ) {
457 if cutoff == 0 || completed.completion_epoch == 0 {
458 return;
459 }
460 for t in self.retry_trackers.lock().iter().filter_map(Weak::upgrade) {
461 let mut l = t.ledger.lock();
462 l.pending.retain(|w| w.strong_count() != 0);
463 for w in l.pending.iter().filter_map(Weak::upgrade) {
464 let f = w.flags.load(Ordering::Acquire);
465 let admission = w.admitted.load(Ordering::Acquire);
466 if w.eligible
467 && w.parent == id
468 && w.birth == completed.completion_epoch
469 && l.fence != 0
470 && w.fence == l.fence
471 && admission != 0
472 && admission < cutoff
473 && f & ADMITTED != 0
474 && f & (STARTED | DISCARDED) == 0
475 && same_contract(id, &completed.data, &w.bytes)
476 {
477 w.flags.fetch_or(RETIRED, Ordering::AcqRel);
478 }
479 }
480 }
481 }
482}
483
484#[cfg(test)]
485mod tests {
486 use super::*;
487 #[test]
488 fn checked_serial_never_wraps_or_reuses_on_exhaustion() {
489 let serial = AtomicU64::new(u64::MAX - 1);
490 assert_eq!(next_serial(&serial), u64::MAX - 1);
491 assert_eq!(next_serial(&serial), 0);
492 assert_eq!(next_serial(&serial), 0);
493 assert_eq!(serial.load(Ordering::Relaxed), u64::MAX);
494 }
495 fn tracker(limits: RetryTrackerLimits) -> RetryTracker {
496 RetryTracker(Arc::new(TrackerInner {
497 router: 1,
498 side: AtomicU64::new(0),
499 limits,
500 ledger: RouterMutex::new(Ledger {
501 fence: 1,
502 pending: Vec::new(),
503 history: Vec::new(),
504 }),
505 }))
506 }
507 #[test]
508 fn opaque_writes_have_no_pruning_or_drain_history() {
509 let tracker = tracker(RetryTrackerLimits::default());
510 let lease = tracker.offer(b"not a canonical packet");
511 assert_eq!(lease.try_start(), WriteStart::PendingAdmission);
512 assert!(lease.report_drained().is_err());
513 lease.admit().unwrap();
514 assert_eq!(lease.try_start(), WriteStart::Ready);
515 lease.report_drained().unwrap();
516 lease.report_drained().unwrap();
517 assert!(tracker.0.ledger.lock().history.is_empty());
518 assert!(tracker.0.ledger.lock().pending.is_empty());
519 assert!(!lease.discard_if_retired());
520 }
521 #[test]
522 fn atomic_group_never_starts_prefix_before_retired_or_unadmitted_follower() {
523 let tracker = tracker(RetryTrackerLimits::default());
524 let first = tracker.offer(b"a");
525 let last = tracker.offer(b"b");
526 first.admit().unwrap();
527 assert_eq!(
528 PackedWriteLease::try_start_batch(&[&first, &last]).unwrap(),
529 WriteStart::PendingAdmission
530 );
531 assert_eq!(first.write.flags.load(Ordering::Relaxed) & STARTED, 0);
532 last.admit().unwrap();
533 last.write.flags.fetch_or(RETIRED, Ordering::Relaxed);
534 assert_eq!(
535 PackedWriteLease::try_start_batch(&[&first, &last]).unwrap(),
536 WriteStart::Retired
537 );
538 assert_eq!(first.write.flags.load(Ordering::Relaxed) & STARTED, 0);
539 tracker.invalidate_generation().unwrap();
540 assert_eq!(
541 PackedWriteLease::try_start_batch(&[&first, &last]).unwrap(),
542 WriteStart::Ready
543 );
544 assert_ne!(first.write.flags.load(Ordering::Relaxed) & STARTED, 0);
545 }
546 #[test]
547 fn group_rejects_duplicates_and_other_transport_instances() {
548 let a = tracker(RetryTrackerLimits::default());
549 let b = tracker(RetryTrackerLimits::default());
550 let x = a.offer(b"a");
551 let y = b.offer(b"b");
552 assert!(PackedWriteLease::try_start_batch(&[&x, &x.clone()]).is_err());
553 assert!(PackedWriteLease::try_start_batch(&[&x, &y]).is_err());
554 assert_eq!(
555 PackedWriteLease::try_start_batch(&[]).unwrap(),
556 WriteStart::Ready
557 );
558 }
559 #[cfg(feature = "discovery")]
560 fn parent(limits: RetryTrackerLimits) -> (Router, RetryTracker, RouterItem, u64) {
561 crate::tests::ensure_common_test_schema();
562 let router =
563 Router::new_with_clock(RouterConfig::new([]).with_sender("GS"), Box::new(|| 0));
564 let (_, tracker) = router
565 .add_side_packed_leased_with_priority_and_options(
566 "radio",
567 |_, _| Ok(()),
568 RouterSideOptions::default(),
569 limits,
570 )
571 .unwrap();
572 let ty = DataType::named("GPS_DATA");
573 let packet = Packet::new_with_wire_contract(
574 ty,
575 &[DataEndpoint::named("RADIO")],
576 "GS",
577 1,
578 1,
579 Arc::from([0u8; 12]),
580 Some(crate::config::get_message_meta(ty).element),
581 vec![Router::sender_hash("FC")].into(),
582 )
583 .unwrap();
584 let id = packet.packet_id();
585 let item = RouterItem::Packet(packet);
586 router
587 .register_end_to_end_reliable_tx(
588 &item,
589 Some(BTreeMap::from([(Router::sender_hash("FC"), 0)])),
590 )
591 .unwrap();
592 (router, tracker, item, id)
593 }
594 #[cfg(feature = "discovery")]
595 fn offered(
596 router: &Router,
597 tracker: &RetryTracker,
598 item: &RouterItem,
599 id: u64,
600 retry: bool,
601 ) -> PackedWriteLease {
602 let _intent = router.retry_intent(Some(id), retry);
603 let RouterItem::Packet(packet) = item else {
604 panic!("packet fixture")
605 };
606 tracker.offer(&wire_format::pack_packet(packet))
607 }
608 #[test]
609 #[cfg(feature = "discovery")]
610 fn admission_after_final_receipt_cutoff_before_scan_is_retained() {
611 let (router, tracker, item, id) = parent(RetryTrackerLimits::default());
612 let first = offered(&router, &tracker, &item, id, false);
613 first.admit().unwrap();
614 assert_eq!(first.try_start(), WriteStart::Ready);
615 first.report_drained().unwrap();
616 let early = offered(&router, &tracker, &item, id, true);
617 early.admit().unwrap();
618 let late = offered(&router, &tracker, &item, id, true);
619 let cutoff = serial();
620 let completed = router
621 .state
622 .lock()
623 .end_to_end_reliable_tx
624 .remove(&id)
625 .unwrap();
626 late.admit().unwrap();
627 router.retire_completed_writes(id, &completed, cutoff);
628 assert!(early.discard_if_retired());
629 assert!(!late.discard_if_retired());
630 assert_eq!(late.try_start(), WriteStart::Ready);
631 }
632 #[test]
633 #[cfg(feature = "discovery")]
634 fn bounded_pending_drop_releases_once_and_drain_history_is_idempotent() {
635 let (router, tracker, item, id) = parent(RetryTrackerLimits {
636 max_pending: 1,
637 ..Default::default()
638 });
639 let first = offered(&router, &tracker, &item, id, false);
640 let clone = first.clone();
641 first.admit().unwrap();
642 assert_eq!(first.try_start(), WriteStart::Ready);
643 first.report_drained().unwrap();
644 first.report_drained().unwrap();
645 assert_eq!(tracker.0.ledger.lock().history.len(), 1);
646 let blocked = offered(&router, &tracker, &item, id, true);
647 assert_eq!(blocked.write.birth, 0);
648 drop(blocked);
649 drop(first);
650 let still_blocked = offered(&router, &tracker, &item, id, true);
651 assert_eq!(still_blocked.write.birth, 0);
652 drop(still_blocked);
653 drop(clone);
654 let retry = offered(&router, &tracker, &item, id, true);
655 assert_ne!(retry.write.birth, 0);
656 assert!(retry.write.eligible);
657 assert_eq!(tracker.0.ledger.lock().pending.len(), 1);
658 }
659 #[test]
660 #[cfg(feature = "discovery")]
661 fn same_bytes_new_logical_send_and_old_generation_drain_are_not_retry_evidence() {
662 let (router, tracker, item, id) = parent(RetryTrackerLimits::default());
663 let first = offered(&router, &tracker, &item, id, false);
664 first.admit().unwrap();
665 assert_eq!(first.try_start(), WriteStart::Ready);
666 first.report_drained().unwrap();
667 let another_first = offered(&router, &tracker, &item, id, false);
668 assert!(!another_first.write.eligible);
669 let retry = offered(&router, &tracker, &item, id, true);
670 assert!(retry.write.eligible);
671 tracker.invalidate_generation().unwrap();
672 let late = offered(&router, &tracker, &item, id, true);
673 assert_eq!(late.write.birth, 0);
674 assert_eq!(retry.try_start(), WriteStart::PendingAdmission);
675 retry.admit().unwrap();
676 assert_eq!(retry.try_start(), WriteStart::Ready);
677 retry.report_drained().unwrap();
678 assert!(tracker.0.ledger.lock().history.is_empty());
679 }
680 #[test]
681 #[cfg(feature = "discovery")]
682 fn changed_contract_or_encoded_bytes_never_match_completion() {
683 let (_router, _tracker, item, id) = parent(RetryTrackerLimits::default());
684 let RouterItem::Packet(p) = &item else {
685 panic!("packet fixture")
686 };
687 let changed = Packet::new_with_wire_contract(
688 p.data_type(),
689 p.endpoints(),
690 p.sender(),
691 p.timestamp(),
692 p.nonce(),
693 Arc::from(p.payload()),
694 p.wire_shape(),
695 vec![Router::sender_hash("OTHER")].into(),
696 )
697 .unwrap();
698 assert_eq!(changed.packet_id(), id);
699 assert!(!same_contract(
700 id,
701 &item,
702 &wire_format::pack_packet(&changed)
703 ));
704 let mut wire = wire_format::pack_packet(p).to_vec();
705 assert!(canonical(&wire));
706 wire[0] |= 0x11;
707 assert!(!canonical(&wire));
708 assert!(!same_contract(id, &item, &wire));
709 }
710}