oxedyne/fe2o3 — fe2o3_ore/src/segment.rs

123 KiB, 446 runs

created by r1870400018:18657, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1	//! An append-only run of operations, as bytes.
2	//!
3	//! A segment is the durable shape of a stretch of history: a short header
4	//! saying what the bytes are, then one length-prefixed record after another,
5	//! each with an integrity check. Appending is writing at the end, and nothing
6	//! already written is ever revisited, so a writer needs no index and a reader
7	//! needs no seek.
8	//!
9	//! Records come in three forms and the format carries all of them, tagged. A
10	//! bare [`Record`] is what a replica writes for itself, where provenance is not
11	//! in question; an [`Envelope`] is the same record with a public key and a
12	//! signature around it, which is what crosses between parties. A segment may
13	//! hold either or both, so a repository that starts unsigned and later gains
14	//! signatures does not need a second format.
15	//!
16	//! The third form is a [`Veiled`] record, which is one of the other two
17	//! encrypted whole, with its header left in clear beside the ciphertext. It
18	//! exists for the case where the machine holding the bytes is not one of the
19	//! parties: a carrier that has to place an operation needs its identifier and
20	//! its parents and nothing else, so those are what it is given.
21	//!
22	//! # The caller owns the cipher too
23	//!
24	//! [`Entry::veil`] and [`Entry::unveil`] take an implementation of [`Encrypter`]
25	//! for the same reason the digest takes a [`Hasher`]: which cipher, and whose
26	//! key, are decisions this crate has no business making. It marshals bytes and
27	//! asks the caller's scheme to encrypt or decrypt them.
28	//!
29	//! # The caller owns the hash
30	//!
31	//! Each record carries a digest, and which function computes it is not decided
32	//! here. The caller brings an implementation of [`Hasher`] and a salt, and the
33	//! same pair must be brought to read the bytes back. That keeps the crate free
34	//! of any particular hash and lets a browser use what its platform offers while
35	//! a server uses what its peers have agreed on.
36	//!
37	//! The digest covers the record's kind byte and its body. It is a check against
38	//! damage, not against forgery: anyone who can rewrite a record can rewrite the
39	//! digest beside it. Forgery is what the signature in an [`Envelope`] is for.
40	//!
41	//! # No I/O
42	//!
43	//! Bytes in, records out. Nothing here opens a file or names a path; a segment
44	//! is a byte buffer that a caller may choose to write to disk.
45	//!
46	//! # Incremental
47	//!
48	//! [`Reader::feed`] takes whatever bytes have arrived and
49	//! [`Reader::next_entry`] yields records as they complete, so a segment larger
50	//! than memory can be read a chunk at a time. Memory is bounded by the largest
51	//! single record rather than by the segment, and feeding a segment one byte at
52	//! a time yields exactly what feeding it all at once yields.
53	//!
54	//! Writing is incremental in the same way, and across runs as well as within
55	//! one: [`Writer::resume`] continues a segment already written and emits only
56	//! the records appended to it, so a caller holding a segment on disk adds to it
57	//! by writing at the end. Resuming reads the existing bytes first, under the
58	//! hasher and salt it is given, so a segment that a later reader could not get
59	//! to the end of is refused before anything is added to it.
60	//!
61	//! [Written with AI entirely](https://need2know.ai/entirely-ai/code)\
62	//! Anthropic Claude
63
64	use crate::envelope::Envelope;
65	use crate::id::{
66	varint_decode,
67	varint_encode,
68	OpId,
69	ReplicaId,
70	VARINT_MAX_LEN,
71	};
72	use crate::op::{
73	Header,
74	Record,
75	};
76
77	use oxedyne_fe2o3_core::prelude::*;
78	use oxedyne_fe2o3_iop_crypto::enc::Encrypter;
79	use oxedyne_fe2o3_iop_hash::api::Hasher;
80	use oxedyne_fe2o3_jdat::prelude::*;
81
82
83	pub const MAGIC: [u8; 6] = *b"ORESEG"; // the bytes every segment begins with
84
85	/// The format version this module writes.
86	///
87	/// Raised to 2 when the operation vocabulary changed for file identity: a
88	/// content operation no longer carries a file, a lifecycle operation names one
89	/// by identity, and a path is bytes rather than a string. Nothing was ever
90	/// written in version 1 that needs to be read again.
91	///
92	/// Raised to 3 when the vocabulary gained [`crate::op::Op::FileMode`] at wire
93	/// code 8. The framing did not move a byte; what the version declares is which
94	/// operations the records inside may be.
95	///
96	/// Raised to 4 when the vocabulary gained five codes at once: wire code 9, the
97	/// second spelling of [`crate::op::Op::Mark`], for a mark carrying what was said
98	/// and when; and wire codes 10 to 13 for [`crate::op::Op::Proposal`],
99	/// [`crate::op::Op::Said`], [`crate::op::Op::Settled`] and
100	/// [`crate::op::Op::Reverts`], which put a proposal, its discussion, its outcome
101	/// and what a revert undoes into history rather than beside it. The framing did
102	/// not move a byte this time either, and neither did a mark that carries neither
103	/// a body nor a time: it is still written at wire code 4 with the two elements it
104	/// always had, so every mark ever signed still verifies.
105	///
106	/// Raised to 5 when the vocabulary gained [`crate::op::Op::Amended`] at wire code
107	/// 14, so that a proposal's author can state it again without the opening
108	/// operation being touched. The framing did not move, and the four proposal codes
109	/// below it did not move either.
110	pub const VERSION: u8 = 6;
111
112	/// The oldest format version this module reads.
113	///
114	/// Version 2 stays readable because each version since has been a strict
115	/// superset of it: the framing is identical and the vocabulary has only ever
116	/// grown upwards, so every version 2 segment ever written means in version 4
117	/// exactly what it meant in version 2, and so does every version 3 one. That is
118	/// what a bump buys -- a reader meeting a version it does not know says which
119	/// version it met, rather than reporting an operation code it cannot place --
120	/// and it is why a bump costs a repository nothing.
121	///
122	/// The rule the vocabulary grows by is what keeps that true, and it is not open
123	/// to reinterpretation: a new code goes strictly above the existing ones,
124	/// [`VERSION`] rises by one, [`highest_code`] gains a branch, and this constant
125	/// stays where it is.
126	///
127	/// Version 1 is not read. Its operations spelled a file as a path and a path as
128	/// a string, so its records are not the same records under another number.
129	pub const VERSION_MIN: u8 = 2;
130
131	/// The highest operation code a segment of the given format version may carry.
132	///
133	/// Version 2 was frozen with [`crate::op::CODE_NOTE`] at the top of the
134	/// vocabulary; version 3 added [`crate::op::Op::FileMode`] above it and nothing
135	/// else; version 4 added the five codes from [`crate::op::CODE_MARK_TIMED`] up to
136	/// [`crate::op::CODE_REVERTS`]; version 5 added [`crate::op::CODE_AMENDED`] above
137	/// those and nothing else. A writer continuing a segment somebody else wrote
138	/// asks this rather than assuming, so that "an older version is a strict subset
139	/// of a newer one" stays true of the bytes and not only of the intention.
140	///
141	/// This is the whole mechanism the additive design rests on. A version 3 segment
142	/// handed an [`crate::op::Op::Reverts`] refuses it here, and the caller starts a
143	/// segment at the current version instead, which is why nothing already written
144	/// has to be rewritten and why nothing already written can be misread.
145	pub const fn highest_code(version: u8) -> u8 {
146	if version <= VERSION_MIN {
147	crate::op::CODE_NOTE
148	} else if version == 3 {
149	crate::op::CODE_FILE_MODE
150	} else if version == 4 {
151	crate::op::CODE_REVERTS
152	} else if version == 5 {
153	crate::op::CODE_AMENDED
154	} else {
155	crate::op::CODE_FORGOTTEN
156	}
157	}
158
159	/// Kind byte of a record carrying a bare [`Record`].
160	pub const KIND_BARE: u8 = 1;
161	/// Kind byte of a record carrying a signed [`Envelope`].
162	pub const KIND_SEALED: u8 = 2;
163	/// Kind byte of a record whose header is in clear and whose body is encrypted.
164	///
165	/// The kind is a separate axis from [`VERSION`], and neither moved for the
166	/// other. A version says which operations the records inside a segment may be; a
167	/// veiled record's operation is ciphertext, so there is no code in it for a
168	/// version to bound, and a segment written at any version this reader knows may
169	/// carry one. What a reader meeting a form it does not have needs to be told is
170	/// the form, and the kind byte says so in the record where it is rather than in a
171	/// header that would condemn every plain record beside it.
172	pub const KIND_VEILED: u8 = 3;
173	/// Kind byte of a record carrying a RUN of records, deflated together.
174	///
175	/// The same axis as [`KIND_VEILED`] and for the same reason: it says what was
176	/// done to the records in the record where they are, rather than in a header
177	/// that would condemn every plain record beside it. [`VERSION`] does not move
178	/// for it, a segment may hold packed and plain records side by side, and a
179	/// reader that meets one and cannot inflate says so by name.
180	///
181	/// A run and not a record. Compression saves what is redundant between
182	/// records, and a record is about 1.4 kB, which is too small a window to see any
183	/// of it: measured on a 55 MB segment, deflating each record on its own reached
184	/// 58.8% where deflating runs of a megabyte reached 38.4%. So one packed record
185	/// carries a run, each run inflatable on its own, and reaching a record costs
186	/// its own megabyte rather than the whole segment.
187	///
188	/// What is inside is the plain framing, unchanged. The inflated bytes are
189	/// exactly the bytes those records would have occupied unpacked, digests
190	/// included, so a packed segment and a plain one carrying the same history yield
191	/// the same records with the same digests in the same order. That is what keeps
192	/// a fold over those digests -- the thing a repack compares two stores by -- the
193	/// same on both sides, and it is why packing is revocable.
194	pub const KIND_PACKED: u8 = 4;
195
196	/// Most bytes a packed run may inflate to.
197	///
198	/// A compressed frame is an instruction to allocate, and the instruction arrives
199	/// from wherever the segment did. The declared run is a megabyte and this is
200	/// sixty-four, so nothing a writer here produces comes near it and a frame that
201	/// does is refused by name rather than obeyed.
202	pub const PACKED_MAX: usize = 64 << 20;
203
204	// How much consumed prefix a reader tolerates before it moves the remainder to
205	// the front of its buffer.
206	const COMPACT_THRESHOLD: usize = 1 << 16;
207
208
209	/// Whether a reader recomputes each record's digest, or takes the one written
210	/// beside it.
211	///
212	/// Recomputing is what a segment's framing is for and it is the default; there
213	/// is no way to reach [`Integrity::Vouched`] except by asking for it. The two
214	/// read the same records out of the same bytes and differ only in what they
215	/// notice about damage.
216	///
217	/// # What vouching costs, and what it is worth
218	///
219	/// The digest covers `[kind] \|\| body` of one record, so recomputing it hashes
220	/// every byte of the segment. Over a 35,408 operation log in 85 MB that is
221	/// 420 ms of a 574 ms decode -- the file I/O beneath it is 7.7 ms -- which is
222	/// paid on every read of a history that has not changed since the last one.
223	///
224	/// What it buys is the only check that survives a segment nothing else looks
225	/// at. A signature says who wrote a record and is skippable by a caller that
226	/// remembers checking it; the digest says the bytes are the bytes, and a body
227	/// byte that flips on the disk changes neither the file's length nor its
228	/// modification time nor the digests recorded beside the bodies. So a caller
229	/// that vouches has stopped looking for bit rot on this read, and something
230	/// else must look for it on some other one. That is not a trade this crate can
231	/// make on a caller's behalf, which is why it is an argument.
232	///
233	/// A caller warranting [`Integrity::Vouched`] warrants three things: that
234	/// these exact bytes were read under [`Integrity::Checked`] at some point, that
235	/// nothing has appended to or rewritten the file since, and that something
236	/// still re-reads them checked on a timescale it has chosen. Vouching for a
237	/// file still being written, or for one this machine has never checked, throws
238	/// the check away and puts nothing in its place.
239	#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
240	pub enum Integrity {
241	#[default]
242	Checked, // hash each body and refuse a record whose digest does not match
243	Vouched, // take the recorded digest as read, on the caller's warrant above
244	}
245
246
247	/// What a segment says about itself before its first record.
248	///
249	/// The replica hint is exactly that: a note of who was writing, which lets a
250	/// reader sort a directory of segments without opening them. Nothing depends on
251	/// it, and a segment whose records come from several replicas simply leaves it
252	/// out rather than lying.
253	#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
254	pub struct Head {
255	pub version: u8, // format version the segment was written in
256	pub replica: Option<ReplicaId>, // who was writing, where one replica wrote all of it
257	}
258
259	impl Head {
260	/// Constructs a header at the current format version.
261	pub fn new(replica: Option<ReplicaId>) -> Self {
262	Self { version: VERSION, replica }
263	}
264
265	/// The shape is the magic, the version, a byte saying whether a replica hint
266	/// follows, and the hint if it does.
267	pub fn encode_into(&self, buf: &mut Vec<u8>) {
268	buf.extend_from_slice(&MAGIC);
269	buf.push(self.version);
270	match self.replica {
271	Some(r) => {
272	buf.push(1);
273	r.encode_into(buf);
274	},
275	None => buf.push(0),
276	}
277	}
278
279	pub fn encode(&self) -> Vec<u8> {
280	let mut buf = Vec::with_capacity(MAGIC.len() + 2 + VARINT_MAX_LEN);
281	self.encode_into(&mut buf);
282	buf
283	}
284
285	/// Yields the header and how many bytes it took. `None` means the bytes so
286	/// far are a prefix of a header and more are needed; an error means they are
287	/// not a header at all.
288	pub fn decode(buf: &[u8])
289	-> Outcome<Option<(Self, usize)>>
290	{
291	if buf.len() < MAGIC.len() {
292	// Only refuse what could not become the magic however it continues.
293	if buf != &MAGIC[..buf.len()] {
294	return Err(err!(
295	"A segment begins {:02x?}, which is not the magic {:02x?}.",
296	buf, MAGIC;
297	Decode, Input, Invalid));
298	}
299	return Ok(None);
300	}
301	if buf[..MAGIC.len()] != MAGIC {
302	return Err(err!(
303	"A segment begins {:02x?}, which is not the magic {:02x?}.",
304	&buf[..MAGIC.len()], MAGIC;
305	Decode, Input, Invalid));
306	}
307	let mut at = MAGIC.len();
308	if buf.len() <= at {
309	return Ok(None);
310	}
311	let version = buf[at];
312	at += 1;
313	if !(VERSION_MIN..=VERSION).contains(&version) {
314	return Err(err!(
315	"A segment declares format version {}, and this reader knows versions \
316	{} to {}.", version, VERSION_MIN, VERSION;
317	Decode, Input, Version, Mismatch));
318	}
319	if buf.len() <= at {
320	return Ok(None);
321	}
322	let tag = buf[at];
323	at += 1;
324	match tag {
325	0 => Ok(Some((Self { version, replica: None }, at))),
326	1 => match res!(try_varint(&buf[at..])) {
327	Some((n, used)) => Ok(Some((
328	Self { version, replica: Some(ReplicaId::new(n)) },
329	at + used,
330	))),
331	None => Ok(None),
332	},
333	other => Err(err!(
334	"A segment's replica hint is tagged {}, which is neither 0 for absent \
335	nor 1 for present.", other;
336	Decode, Input, Invalid)),
337	}
338	}
339	}
340
341
342	/// An entry whose header can be read and whose body cannot.
343	///
344	/// The header is in clear because a carrier that never reads an operation still
345	/// has to place one: a frontier walk follows parents, a sketch is keyed by
346	/// identifiers, and a closure check wants both. None of them wants an operation
347	/// body, which is what rendering wants, and a carrier does not render. So this is
348	/// the whole of what a repository gives away to be carried, and the rest of it is
349	/// ciphertext.
350	///
351	/// The signature is inside, over the plaintext record, which puts verification
352	/// where decryption is: at a reader holding the key, never at the carrier. The
353	/// clear header duplicates the one sealed inside, and that duplication is what
354	/// makes a carrier that alters it detectable rather than merely suspected --
355	/// [`Entry::unveil`] compares the two and refuses the pair if they disagree.
356	#[derive(Clone, Debug, Eq, PartialEq)]
357	pub struct Veiled {
358	pub head: Header, // identifier and parents, in clear
359	pub body: Vec<u8>, // the whole entry, encrypted
360	}
361
362	impl Veiled {
363	/// The shape is `[head, body]`.
364	pub fn to_dat(&self) -> Dat {
365	Dat::List(vec![
366	self.head.to_dat(),
367	Dat::BU64(self.body.clone()),
368	])
369	}
370
371	pub fn from_dat(dat: &Dat)
372	-> Outcome<Self>
373	{
374	let v = match dat {
375	Dat::List(v) if v.len() == 2 => v,
376	_ => return Err(err!(
377	"A veiled record expects a 2-element Dat::List, got {:?}.", dat;
378	Decode, Input, Mismatch)),
379	};
380	let head = res!(Header::from_dat(&v[0]));
381	// One width and not the narrower ones a shorter body would also fit. Two
382	// byte spellings of one veiled entry would both decode to the same thing
383	// and hash differently, and a record whose digest depends on which
384	// spelling it arrived in is not one a carrier can pass on unaltered.
385	let body = match &v[1] {
386	Dat::BU64(b) => b.clone(),
387	other => return Err(err!(
388	"The veiled body of {} is encoded {:?}; it is written under a 64-bit \
389	length, whatever its size.", head.id(), other;
390	Decode, Input, Mismatch)),
391	};
392	Ok(Self { head, body })
393	}
394	}
395
396
397	/// One record of a segment: an operation, with or without its provenance, and
398	/// readable or not.
399	#[derive(Clone, Debug, Eq, PartialEq)]
400	pub enum Entry {
401	/// An operation written down as it stands.
402	Bare(Record),
403	/// An operation with a public key and a signature around it.
404	Sealed(Envelope),
405	/// An operation whose header is in clear and whose body is encrypted.
406	Veiled(Veiled),
407	}
408
409	impl Entry {
410	/// Returns the kind byte identifying the form.
411	pub fn kind(&self) -> u8 {
412	match self {
413	Self::Bare(_) => KIND_BARE,
414	Self::Sealed(_) => KIND_SEALED,
415	Self::Veiled(_) => KIND_VEILED,
416	}
417	}
418
419	/// The form's name, for messages.
420	pub fn name(&self) -> &'static str {
421	match self {
422	Self::Bare(_) => "bare record",
423	Self::Sealed(_) => "sealed envelope",
424	Self::Veiled(_) => "veiled record",
425	}
426	}
427
428	pub fn is_veiled(&self) -> bool {
429	matches!(self, Self::Veiled(_))
430	}
431
432	/// The header, which every form carries in clear.
433	///
434	/// This is the one question a carrier may ask of any entry whatever, and it is
435	/// what separates carrying a history from reading one. A veiled entry answers
436	/// from the clear header beside its ciphertext and the other two from the
437	/// record they hold, so a caller that wants the graph and not the content never
438	/// wants a key.
439	pub fn head(&self)
440	-> Outcome<Header>
441	{
442	Ok(match self {
443	Self::Bare(rec) => rec.head.clone(),
444	Self::Sealed(e) => res!(e.peek_record()).head,
445	Self::Veiled(v) => v.head.clone(),
446	})
447	}
448
449	/// Opens a sealed record without checking its signature.
450	///
451	/// Verification is the caller's to do, with the scheme the caller holds; a
452	/// segment reader has no key material and makes no claim about provenance.
453	///
454	/// A veiled entry fails here rather than answering with a stand-in, because
455	/// every caller of this asks it in order to read an operation, and a body
456	/// nobody can read is not one. The failure names the operation, so a carrier
457	/// handed a form it was not built for says which one it was.
458	pub fn peek(&self)
459	-> Outcome<Record>
460	{
461	match self {
462	Self::Bare(rec) => Ok(rec.clone()),
463	Self::Sealed(e) => e.peek_record(),
464	Self::Veiled(v) => Err(err!(
465	"The operation {} is veiled: its body is encrypted under a key held by \
466	whoever may read this repository, and not by whoever carries it. Its \
467	header is readable with `Entry::head`, and its body with \
468	`Entry::unveil` and the key.", v.head.id();
469	Invalid, Input, Missing, Key)),
470	}
471	}
472
473	pub fn id(&self)
474	-> Outcome<OpId>
475	{
476	Ok(res!(self.head()).id())
477	}
478
479	/// Encrypts an entry whole, leaving its header in clear.
480	///
481	/// What goes under the cipher is the entry's own tagged form, so the form
482	/// travels with it: a sealed envelope unveils to a sealed envelope, signature
483	/// and public key intact, and a bare record to a bare record. Nothing about the
484	/// entry is re-encoded on the way, so a signature made before it was veiled is
485	/// the signature checked after it is unveiled.
486	///
487	/// Veiling a veiled entry is refused. A second wrapping would hide a header
488	/// that is already hidden, which is the one thing the form exists not to do.
489	pub fn veil<E: Encrypter>(&self, enc: &E)
490	-> Outcome<Self>
491	{
492	if let Self::Veiled(v) = self {
493	return Err(err!(
494	"The operation {} is veiled already, and veiling it again would hide \
495	the header a carrier places it by.", v.head.id();
496	Invalid, Input, Duplicate));
497	}
498	let head = res!(self.head());
499	let plain = res!(self.to_dat().to_bytes(Vec::new()));
500	Ok(Self::Veiled(Veiled { head, body: res!(enc.encrypt(&plain)) }))
501	}
502
503	/// Decrypts a veiled entry, and refuses one whose clear header is not the
504	/// header inside it.
505	///
506	/// The comparison is the whole of what the duplicated header buys. A carrier
507	/// cannot touch the copy inside, which is under the signature, but it can
508	/// rewrite the copy in clear, and every peer that never holds the key would
509	/// place the operation by the rewritten one. So the first reader with the key
510	/// checks the two against each other, and a disagreement is refused by name
511	/// with the signed copy named as the one to believe.
512	pub fn unveil<E: Encrypter>(&self, enc: &E)
513	-> Outcome<Self>
514	{
515	let veiled = match self {
516	Self::Veiled(v) => v,
517	other => return Err(err!(
518	"A {} is not veiled, so there is nothing to unveil.", other.name();
519	Invalid, Input, Mismatch)),
520	};
521	let plain = match enc.decrypt(&veiled.body) {
522	Ok(p) => p,
523	Err(e) => return Err(err!(e,
524	"The {} byte body of the veiled operation {} did not decrypt. Either \
525	this is not the key the repository was veiled under, or the bytes have \
526	been altered since.", veiled.body.len(), veiled.head.id();
527	Invalid, Input, Decrypt, Key)),
528	};
529	let (dat, used) = res!(Dat::from_bytes(&plain));
530	if used != plain.len() {
531	return Err(err!(
532	"The veiled operation {} decrypted to {} bytes and decoded from only {} \
533	of them.", veiled.head.id(), plain.len(), used;
534	Decode, Input, Mismatch));
535	}
536	let inner = res!(Self::from_dat(&dat));
537	if inner.is_veiled() {
538	return Err(err!(
539	"The veiled operation {} holds another veiled record.", veiled.head.id();
540	Decode, Input, Invalid));
541	}
542	let inside = res!(inner.head());
543	if inside != veiled.head {
544	return Err(err!(
545	"A veiled entry says in clear that it is {} written against {}, and the \
546	record inside it is {} written against {}. The clear header is what a \
547	carrier places an operation by, so the two disagreeing means the carrier \
548	was given one history and shown another; the record inside is the signed \
549	one and is what to believe.",
550	veiled.head.id(), said_parents(&veiled.head),
551	inside.id(), said_parents(&inside);
552	Invalid, Input, Security, Mismatch));
553	}
554	Ok(inner)
555	}
556
557	/// What the entry comes to in a carrier, without any of it being encoded.
558	///
559	/// Exactly the length [`Entry::to_dat`] encodes to, and which form that is
560	/// matters: this is the tagged shape a sync message carries, not the untagged
561	/// body a segment writes beside a kind byte of its own.
562	///
563	/// A carrier that bounds what it will send measures every entry it considers
564	/// and sends only some of them, so measuring by serialising buys a number at
565	/// the price of the history it is about to throw away. On fe2o3's own history
566	/// that is a 22,153,680 byte operation encoded and discarded once per clone.
567	pub fn dat_len(&self)
568	-> Outcome<usize>
569	{
570	Ok(res!(self.to_dat().byte_len().ok_or_else(\|\| err!(
571	"A {} holds a daticle whose encoded length cannot be known without \
572	encoding it, which is a kind no entry was ever built to carry.", self.name();
573	Bug, Invalid))))
574	}
575
576	/// The shape is `[kind, body]`.
577	///
578	/// This is the form for a carrier that is itself a daticle, such as a sync
579	/// message. A segment does not use it: there the kind is a byte of the frame
580	/// and the body stands alone, so that the digest can cover both without
581	/// re-encoding.
582	pub fn to_dat(&self) -> Dat {
583	Dat::List(vec![
584	Dat::U8(self.kind()),
585	match self {
586	Self::Bare(rec) => rec.to_dat(),
587	Self::Sealed(e) => e.to_dat(),
588	Self::Veiled(v) => v.to_dat(),
589	},
590	])
591	}
592
593	pub fn from_dat(dat: &Dat)
594	-> Outcome<Self>
595	{
596	let v = match dat {
597	Dat::List(v) if v.len() == 2 => v,
598	_ => return Err(err!(
599	"An Entry expects a 2-element Dat::List, got {:?}.", dat;
600	Decode, Input, Mismatch)),
601	};
602	let kind = match &v[0] {
603	Dat::U8(k) => *k,
604	other => return Err(err!(
605	"An Entry kind expects Dat::U8, got {:?}.", other;
606	Decode, Input, Mismatch)),
607	};
608	match kind {
609	KIND_BARE => Ok(Self::Bare(res!(Record::from_dat(&v[1])))),
610	KIND_SEALED => Ok(Self::Sealed(res!(Envelope::from_dat(&v[1])))),
611	KIND_VEILED => Ok(Self::Veiled(res!(Veiled::from_dat(&v[1])))),
612	other => Err(err!(
613	"An Entry is tagged {}, which is none of {} for a bare record, {} for a \
614	sealed envelope and {} for a veiled one.",
615	other, KIND_BARE, KIND_SEALED, KIND_VEILED;
616	Decode, Input, Invalid)),
617	}
618	}
619
620	/// The daticle form of whichever shape the entry holds.
621	pub fn body(&self)
622	-> Outcome<Vec<u8>>
623	{
624	let dat = match self {
625	Self::Bare(rec) => rec.to_dat(),
626	Self::Sealed(e) => e.to_dat(),
627	Self::Veiled(v) => v.to_dat(),
628	};
629	Ok(res!(dat.to_bytes(Vec::new())))
630	}
631
632	fn from_body(kind: u8, body: &[u8])
633	-> Outcome<Self>
634	{
635	let (dat, used) = res!(Dat::from_bytes(body));
636	if used != body.len() {
637	return Err(err!(
638	"A segment record body of {} bytes decoded from only {} of them.",
639	body.len(), used;
640	Decode, Input, Mismatch));
641	}
642	match kind {
643	KIND_BARE => Ok(Self::Bare(res!(Record::from_dat(&dat)))),
644	KIND_SEALED => Ok(Self::Sealed(res!(Envelope::from_dat(&dat)))),
645	KIND_VEILED => Ok(Self::Veiled(res!(Veiled::from_dat(&dat)))),
646	other => Err(err!(
647	"A segment record is tagged {}, which is none of {} for a bare record, \
648	{} for a sealed envelope and {} for a veiled one.",
649	other, KIND_BARE, KIND_SEALED, KIND_VEILED;
650	Decode, Input, Invalid)),
651	}
652	}
653	}
654
655
656	/// Builds a segment, record by record.
657	///
658	/// The bytes accumulate in memory; where they go afterwards is the caller's
659	/// business. A writer is generic over the hasher rather than taking one per
660	/// call, so that every record of a segment is checked the same way by
661	/// construction.
662	///
663	/// A writer either starts a segment, with [`Writer::new`], or continues one
664	/// already written, with [`Writer::resume`]. The difference is only whether the
665	/// header is emitted, since a segment is its header and then records to the end;
666	/// what a resumed writer hands back is the records alone, to be appended to the
667	/// bytes they continue.
668	#[derive(Clone, Debug)]
669	pub struct Writer<H: Hasher, const S: usize> {
670	hasher: H, // hash function each record's digest is computed with
671	salt: [u8; S], // salt each digest is computed under
672	version: u8, // declared format version, which bounds the vocabulary
673	buf: Vec<u8>, // bytes written so far
674	count: usize, // records held, those resumed from included
675	}
676
677	impl<H: Hasher, const S: usize> Writer<H, S> {
678
679	/// Constructs a writer, emitting the segment header at once.
680	pub fn new(head: &Head, hasher: H, salt: [u8; S]) -> Self {
681	let mut buf = Vec::new();
682	head.encode_into(&mut buf);
683	Self { hasher, salt, version: head.version, buf, count: 0 }
684	}
685
686	/// The bytes handed back afterwards are the new records alone, which a caller
687	/// appends to the segment they were resumed from; the header is not emitted a
688	/// second time. [`Writer::count`] carries on from the records already there.
689	///
690	/// `existing` is read through first, under the hasher and the salt given, and
691	/// that is what makes appending safe: a different hash function, a different
692	/// salt, a segment written in another format version, and a segment left
693	/// half-written by an interrupted append all fail here, rather than being
694	/// quietly extended into bytes no reader can get to the end of.
695	pub fn resume(existing: &[u8], hasher: H, salt: [u8; S])
696	-> Outcome<Self>
697	{
698	let mut reader: Reader<H, S> = Reader::new(hasher.clone(), salt);
699	reader.feed(existing);
700	reader.end();
701	// Every record is decoded and its digest checked, and nothing is kept:
702	// what is wanted is the count and the assurance, not the operations.
703	while res!(reader.next_entry()).is_some() {}
704	let version = match reader.head() {
705	Some(head) => head.version,
706	None => return Err(err!(
707	"A segment of {} bytes carries no header, so there is nothing to \
708	continue.", existing.len();
709	Decode, Input, Missing)),
710	};
711	Ok(Self { hasher, salt, version, buf: Vec::new(), count: reader.count() })
712	}
713
714	pub const fn version(&self) -> u8 {
715	self.version
716	}
717
718	/// A record is refused where the segment's declared version has no code for
719	/// the operation it carries, which is what keeps an older version a genuine
720	/// subset of a newer one rather than a promise the bytes break. A caller with
721	/// such a record to write starts a segment at the current version instead;
722	/// nothing about a log says its segments share a version.
723	///
724	/// A veiled record is not asked, because nothing here can ask it: its
725	/// operation is ciphertext and carries no code for the version to bound. That
726	/// is not a hole in the claim, since the claim is about what a reader of the
727	/// segment can be handed, and what a reader is handed here is an opaque body
728	/// and the name of a key it does not have.
729	pub fn push(&mut self, entry: &Entry)
730	-> Outcome<()>
731	{
732	res!(self.admits(entry));
733	let mut framed = Vec::new();
734	res!(self.frame_into(entry, &mut framed));
735	self.buf.extend_from_slice(&framed);
736	self.count += 1;
737	Ok(())
738	}
739
740	/// Writes a run of entries as one packed record.
741	///
742	/// What goes under the compressor is the framing those entries would have had
743	/// written plainly -- kind, length, body, digest, one after another -- so
744	/// inflating yields exactly the bytes a plain segment holds and the records
745	/// come back with the digests they always had. Nothing about an entry is
746	/// re-encoded on the way in or out, so a signature made before packing is the
747	/// signature checked after it.
748	///
749	/// The outer record carries a digest of the compressed bytes, which is what
750	/// catches damage before anything is inflated.
751	pub fn push_packed(&mut self, entries: &[Entry])
752	-> Outcome<()>
753	{
754	if entries.is_empty() {
755	return Err(err!(
756	"A packed record was asked for over no entries. An empty run would be \
757	a record carrying nothing, which a reader cannot tell from a damaged \
758	one."; Invalid, Input, Missing));
759	}
760	let mut plain = Vec::new();
761	for entry in entries {
762	res!(self.admits(entry));
763	res!(self.frame_into(entry, &mut plain));
764	}
765	let body = res!(deflate(&plain));
766	let digest = self.hasher.clone().hash(&[&[KIND_PACKED], &body], self.salt).as_vec();
767	self.buf.push(KIND_PACKED);
768	varint_encode(body.len() as u64, &mut self.buf);
769	self.buf.extend_from_slice(&body);
770	varint_encode(digest.len() as u64, &mut self.buf);
771	self.buf.extend_from_slice(&digest);
772	self.count += entries.len();
773	Ok(())
774	}
775
776	/// Refuses an entry whose operation the declared version has no code for.
777	fn admits(&self, entry: &Entry)
778	-> Outcome<()>
779	{
780	if self.version < VERSION && !entry.is_veiled() {
781	let op = res!(entry.peek()).op;
782	let top = highest_code(self.version);
783	if op.code() > top {
784	return Err(err!(
785	"A segment declaring format version {} cannot carry an {}, whose \
786	wire code {} is above the {} that version spells; version {} is \
787	where that operation was added.",
788	self.version, op.name(), op.code(), top, VERSION;
789	Invalid, Input, Version, Mismatch));
790	}
791	}
792	Ok(())
793	}
794
795	/// Appends one record's framing, which is the same whether it is going
796	/// straight into the segment or into a run about to be packed.
797	fn frame_into(&self, entry: &Entry, out: &mut Vec<u8>)
798	-> Outcome<()>
799	{
800	let kind = entry.kind();
801	let body = res!(entry.body());
802	let digest = self.hasher.clone().hash(&[&[kind], &body], self.salt).as_vec();
803	out.push(kind);
804	varint_encode(body.len() as u64, out);
805	out.extend_from_slice(&body);
806	varint_encode(digest.len() as u64, out);
807	out.extend_from_slice(&digest);
808	Ok(())
809	}
810
811	pub fn extend<'a, I>(&mut self, entries: I)
812	-> Outcome<()>
813	where
814	I: IntoIterator<Item = &'a Entry>,
815	{
816	for entry in entries {
817	res!(self.push(entry));
818	}
819	Ok(())
820	}
821
822	/// Counting the records a resumed writer was given as well as those it has
823	/// written.
824	pub fn count(&self) -> usize {
825	self.count
826	}
827
828	/// For a resumed writer, the new records alone.
829	pub fn bytes(&self) -> &[u8] {
830	&self.buf
831	}
832
833	/// For a resumed writer, the bytes to append to the segment it continues.
834	pub fn finish(self) -> Vec<u8> {
835	self.buf
836	}
837	}
838
839
840	/// Reads a segment as its bytes arrive.
841	///
842	/// Bytes go in through [`Reader::feed`], records come out through
843	/// [`Reader::next_entry`], and [`Reader::end`] declares that no more bytes are
844	/// coming. Until `end` has been called, `next_entry` returning `None` means only
845	/// that the next record is not yet complete; afterwards it means the segment is
846	/// finished, and a record left half-written is an error.
847	#[derive(Clone, Debug)]
848	pub struct Reader<H: Hasher, const S: usize> {
849	hasher: H, // hash function each record's digest is checked with
850	salt: [u8; S], // salt each digest is checked under
851	buf: Vec<u8>, // bytes fed but not turned into records, consumed prefix included
852	pos: usize, // how much of `buf` has been consumed
853	eof: bool, // whether the caller has declared the segment complete
854	head: Option<Head>, // the header, once it has been read
855	count: usize, // records handed over
856	tally: Option<Vec<u8>>, // digests of records handed over since the last take
857	check: Integrity, // whether each record's digest is recomputed
858	// A packed record inflated, and how much of it has been handed over. A run is
859	// drained before another byte of the segment is looked at, so the records
860	// come out in the order they went in and a caller cannot tell a packed
861	// segment from a plain one.
862	run: Vec<u8>,
863	ran: usize,
864	}
865
866	impl<H: Hasher, const S: usize> Reader<H, S> {
867
868	pub fn new(hasher: H, salt: [u8; S]) -> Self {
869	Self {
870	hasher,
871	salt,
872	buf: Vec::new(),
873	pos: 0,
874	eof: false,
875	head: None,
876	count: 0,
877	tally: None,
878	check: Integrity::Checked,
879	run: Vec::new(),
880	ran: 0,
881	}
882	}
883
884	/// Reads on the caller's warrant that these bytes have already been checked,
885	/// rather than checking them again.
886	///
887	/// Read [`Integrity`] before reaching for this. It takes the segment's only
888	/// defence against a body byte that flipped on the disk out of the read, and
889	/// it is the caller who has to say where that defence went instead.
890	pub fn integrity(mut self, check: Integrity) -> Self {
891	self.check = check;
892	self
893	}
894
895	/// A reader that also keeps the digest of every record it hands over, for a
896	/// caller that wants to name the exact bytes it read.
897	///
898	/// Each record's digest is computed to check it and then dropped, so a caller
899	/// that wanted one had no way to ask and would have to hash the segment a
900	/// second time -- 205 ms over a 55 MB segment, measured, against the 7 ms of
901	/// hashing digests that were paid for already. Take them with
902	/// [`Reader::take_digests`] as they accumulate, which is what keeps them from
903	/// growing to one digest for every record in the segment.
904	pub fn tallying(hasher: H, salt: [u8; S]) -> Self {
905	let mut reader = Self::new(hasher, salt);
906	reader.tally = Some(Vec::new());
907	reader
908	}
909
910	/// Takes up a segment part way through, at a byte that is a record boundary.
911	///
912	/// A reader ordinarily learns the header from the bytes it is fed, and the
913	/// only place a header is written is the first bytes of the segment. So a
914	/// reader that is to be fed from the middle has to be told two things the
915	/// bytes it will see do not carry: the header the segment declared, and how
916	/// many records stand before the first one it will be handed. The header is
917	/// what lets it place records at all; the count is what lets a damaged record
918	/// name its own position in the file rather than its position in the read.
919	///
920	/// The caller warrants that the next byte fed begins a record. Nothing
921	/// here can check that: a segment carries no index and a record is found only
922	/// by reading the one before it, so the only party that knows where a record
923	/// begins is the read that stopped there. A byte offset taken from anywhere
924	/// else will be refused as a damaged record, which is the right answer given
925	/// the wrong question.
926	///
927	/// A segment grows only at its end and nothing already written is ever
928	/// revisited, which is what makes taking one up worth doing: a reader that
929	/// kept where it stopped reads only what has arrived since.
930	pub fn take_up(&mut self, head: Head, ordinal: usize) {
931	self.head = Some(head);
932	self.count = ordinal;
933	}
934
935	/// The digests of the records handed over since the last call, in order, for
936	/// a reader built by [`Reader::tallying`]. Empty for any other.
937	pub fn take_digests(&mut self) -> Vec<u8> {
938	match &mut self.tally {
939	Some(tally) => std::mem::take(tally),
940	None => Vec::new(),
941	}
942	}
943
944	/// Chunk boundaries carry no meaning: a record may be split anywhere, and
945	/// the same segment delivered in different chunkings yields the same
946	/// records.
947	pub fn feed(&mut self, chunk: &[u8]) {
948	self.buf.extend_from_slice(chunk);
949	}
950
951	/// Declares that no further bytes will be fed.
952	pub fn end(&mut self) {
953	self.eof = true;
954	}
955
956	/// `None` until enough bytes have arrived for the header to be read.
957	pub fn head(&self)
958	-> Option<&Head>
959	{
960	self.head.as_ref()
961	}
962
963	pub fn count(&self) -> usize {
964	self.count
965	}
966
967	pub fn remaining(&self) -> &[u8] {
968	&self.buf[self.pos..]
969	}
970
971	/// Has the segment ended with every byte of it turned into a record?
972	pub fn is_exhausted(&self) -> bool {
973	self.eof && self.pos >= self.buf.len() && self.ran >= self.run.len()
974	}
975
976	/// `None` means that the next record is not yet complete, or, once
977	/// [`Reader::end`] has been called, that the segment is finished. An error
978	/// names the record that could not be read.
979	pub fn next_entry(&mut self)
980	-> Outcome<Option<Entry>>
981	{
982	if self.head.is_none() {
983	match res!(Head::decode(&self.buf[self.pos..])) {
984	Some((head, used)) => {
985	self.head = Some(head);
986	self.pos += used;
987	self.compact();
988	},
989	None => {
990	if self.eof {
991	return Err(err!(
992	"A segment ends part way through its header, after {} \
993	byte{}.", self.buf.len() - self.pos,
994	if self.buf.len() - self.pos == 1 { "" } else { "s" };
995	Decode, Input, Missing));
996	}
997	return Ok(None);
998	},
999	}
1000	}
1001	// A run already inflated is drained first, so that the records of a packed
1002	// segment arrive in the order they were packed and the caller cannot tell
1003	// which kind of segment it is reading.
1004	if self.ran < self.run.len() {
1005	let (entry, used, digest) = {
1006	let taken = res!(framed(
1007	&self.hasher, self.salt, &self.run[self.ran..], self.count, self.check));
1008	match taken {
1009	Some(f) => {
1010	if f.kind == KIND_PACKED {
1011	return Err(err!(
1012	"Record {} of a packed run is itself packed. A run holds \
1013	the records it packed and nothing else; a run inside a run \
1014	would hide the framing a reader places records by.",
1015	self.count;
1016	Decode, Input, Invalid));
1017	}
1018	(res!(Entry::from_body(f.kind, f.body)), f.used, f.digest.to_vec())
1019	},
1020	None => return Err(err!(
1021	"A packed run ends part way through record {}, with {} byte{} \
1022	left over. The run inflated and what came out is not the framing \
1023	that went in.", self.count, self.run.len() - self.ran,
1024	if self.run.len() - self.ran == 1 { "" } else { "s" };
1025	Decode, Input, Missing)),
1026	}
1027	};
1028	self.ran += used;
1029	self.count += 1;
1030	if let Some(tally) = &mut self.tally {
1031	tally.extend_from_slice(&digest);
1032	}
1033	if self.ran >= self.run.len() {
1034	self.run = Vec::new();
1035	self.ran = 0;
1036	}
1037	return Ok(Some(entry));
1038	}
1039	if self.pos >= self.buf.len() {
1040	return Ok(None);
1041	}
1042	// The outer digest of a packed record covers the compressed bytes and is
1043	// checked before anything is inflated, which is what stops a damaged frame
1044	// becoming an instruction to allocate. It is NOT tallied: what a fold over
1045	// a log names is the records, and a packed segment holds the same records
1046	// as the plain one it was made from.
1047	let made = {
1048	let taken = res!(framed(
1049	&self.hasher, self.salt, &self.buf[self.pos..], self.count, self.check));
1050	match taken {
1051	Some(f) if f.kind == KIND_PACKED => Some((Made::Run(res!(
1052	inflate(f.body, self.count))), f.used)),
1053	Some(f) => Some((Made::One(res!(
1054	Entry::from_body(f.kind, f.body)), f.digest.to_vec()), f.used)),
1055	None => None,
1056	}
1057	};
1058	match made {
1059	Some((Made::One(entry, digest), used)) => {
1060	self.pos += used;
1061	self.count += 1;
1062	if let Some(tally) = &mut self.tally {
1063	tally.extend_from_slice(&digest);
1064	}
1065	self.compact();
1066	Ok(Some(entry))
1067	},
1068	Some((Made::Run(run), used)) => {
1069	self.pos += used;
1070	self.run = run;
1071	self.ran = 0;
1072	self.compact();
1073	self.next_entry()
1074	},
1075	None => {
1076	if self.eof {
1077	Err(err!(
1078	"A segment ends part way through record {}, with {} byte{} \
1079	left over.", self.count, self.buf.len() - self.pos,
1080	if self.buf.len() - self.pos == 1 { "" } else { "s" };
1081	Decode, Input, Missing))
1082	} else {
1083	Ok(None)
1084	}
1085	},
1086	}
1087	}
1088
1089	/// Drops the consumed prefix of the buffer once it is worth the move.
1090	fn compact(&mut self) {
1091	if self.pos == self.buf.len() {
1092	self.buf.clear();
1093	self.pos = 0;
1094	} else if self.pos >= COMPACT_THRESHOLD {
1095	self.buf.drain(..self.pos);
1096	self.pos = 0;
1097	}
1098	}
1099	}
1100
1101
1102	/// What one step of [`Reader::next_entry`] produced from the segment: a record,
1103	/// or a run to be handed over a record at a time.
1104	enum Made {
1105	One(Entry, Vec<u8>), // the entry, and the digest it was checked against
1106	Run(Vec<u8>), // a packed record inflated, still framed
1107	}
1108
1109	/// One framed record, with the digest written beside it.
1110	///
1111	/// The kind is handed back rather than interpreted, because what is done next
1112	/// depends on it: a bare, sealed or veiled record becomes an [`Entry`], and a
1113	/// packed one becomes a run of them.
1114	struct Framed<'a> {
1115	kind: u8,
1116	body: &'a [u8],
1117	used: usize, // bytes of `buf` the whole record occupied
1118	digest: &'a [u8], // the record's digest, which is what a fold wants
1119	}
1120
1121	/// Reads one framed record from the front of `buf`, whether that is a segment
1122	/// being fed or a run just inflated.
1123	///
1124	/// `None` means the bytes so far are a prefix of a record and more are needed;
1125	/// the caller decides whether more can arrive. `ordinal` is only for the
1126	/// messages, so that a damaged record names its own position.
1127	///
1128	/// Under [`Integrity::Vouched`] the body is not hashed and the recorded digest
1129	/// is handed back unexamined, so what comes out of a run of records is the same
1130	/// bytes either way and only the damage a fold could name differs.
1131	fn framed<'a, H: Hasher, const S: usize>(
1132	hasher: &H,
1133	salt: [u8; S],
1134	buf: &'a [u8],
1135	ordinal: usize,
1136	check: Integrity,
1137	)
1138	-> Outcome<Option<Framed<'a>>>
1139	{
1140	if buf.is_empty() {
1141	return Ok(None);
1142	}
1143	let kind = buf[0];
1144	let mut at = 1usize;
1145	let (len, used) = match res!(try_varint(&buf[at..])) {
1146	Some(v) => v,
1147	None => return Ok(None),
1148	};
1149	at += used;
1150	let body_end = match at.checked_add(len as usize) {
1151	Some(e) if (len as u64) <= usize::MAX as u64 => e,
1152	_ => return Err(err!(
1153	"Record {} of the segment declares a body of {} bytes, which no buffer \
1154	can hold.", ordinal, len;
1155	Decode, Input, Excessive)),
1156	};
1157	if buf.len() < body_end {
1158	return Ok(None);
1159	}
1160	let body = &buf[at..body_end];
1161	at = body_end;
1162	let (dlen, used) = match res!(try_varint(&buf[at..])) {
1163	Some(v) => v,
1164	None => return Ok(None),
1165	};
1166	at += used;
1167	let digest_end = match at.checked_add(dlen as usize) {
1168	Some(e) if (dlen as u64) <= usize::MAX as u64 => e,
1169	_ => return Err(err!(
1170	"Record {} of the segment declares a digest of {} bytes, which no buffer \
1171	can hold.", ordinal, dlen;
1172	Decode, Input, Excessive)),
1173	};
1174	if buf.len() < digest_end {
1175	return Ok(None);
1176	}
1177	let digest = &buf[at..digest_end];
1178	if let Integrity::Vouched = check {
1179	return Ok(Some(Framed { kind, body, used: digest_end, digest }));
1180	}
1181	let want = hasher.clone().hash(&[&[kind], body], salt).as_vec();
1182	if want != digest {
1183	// Naming the operation is worth a decode attempt, since a caller with a
1184	// damaged segment wants to know which edit is at risk. Where the body is
1185	// too far gone to decode, the ordinal is all there is to say.
1186	let named = match kind {
1187	KIND_PACKED => fmt!("a run of packed operations"),
1188	_ => match Entry::from_body(kind, body) {
1189	Ok(entry) => match entry.id() {
1190	Ok(id) => fmt!("the operation {}", id),
1191	Err(_) => fmt!("an unreadable operation"),
1192	},
1193	Err(_) => fmt!("an unreadable operation"),
1194	},
1195	};
1196	return Err(err!(
1197	"Record {} of the segment, carrying {}, fails its integrity check: {} \
1198	bytes of body hash to {:02x?}, and {:02x?} was recorded.",
1199	ordinal, named, body.len(), want, digest;
1200	Decode, Input, Checksum, Mismatch));
1201	}
1202	Ok(Some(Framed { kind, body, used: digest_end, digest }))
1203	}
1204
1205	/// Compresses a run's plain framing.
1206	fn deflate(plain: &[u8])
1207	-> Outcome<Vec<u8>>
1208	{
1209	let mut out = Vec::new();
1210	let mut enc = flate2::write::DeflateEncoder::new(&mut out, flate2::Compression::new(6));
1211	match std::io::Write::write_all(&mut enc, plain) {
1212	Ok(()) => (),
1213	Err(e) => return Err(err!(e,
1214	"{} bytes of records could not be compressed.", plain.len();
1215	Encode, Data)),
1216	}
1217	match enc.finish() {
1218	Ok(_) => (),
1219	Err(e) => return Err(err!(e,
1220	"{} bytes of records could not be compressed.", plain.len();
1221	Encode, Data)),
1222	}
1223	Ok(out)
1224	}
1225
1226	/// Inflates a packed run, refusing one that would not stop.
1227	///
1228	/// A compressed frame is an instruction to allocate and it arrives from wherever
1229	/// the segment did, so the output is bounded by [`PACKED_MAX`] and a frame that
1230	/// reaches it is refused rather than obeyed. The bound is what makes
1231	/// [`Integrity::Vouched`] safe to offer: under [`Integrity::Checked`] the digest
1232	/// over the compressed bytes has held by the time this is called and what
1233	/// remains to guard against is a frame somebody wrote to be obeyed, but a
1234	/// vouched read reaches here with the frame unexamined and the same bound holds
1235	/// it.
1236	fn inflate(body: &[u8], ordinal: usize)
1237	-> Outcome<Vec<u8>>
1238	{
1239	let mut out = Vec::new();
1240	// `std::io::Read::take`, not the iterator's: the bound is on bytes read.
1241	let mut dec = std::io::Read::take(
1242	flate2::read::DeflateDecoder::new(body), (PACKED_MAX as u64) + 1);
1243	match std::io::Read::read_to_end(&mut dec, &mut out) {
1244	Ok(_) => (),
1245	Err(e) => return Err(err!(e,
1246	"The packed run at record {} carries {} bytes that do not inflate. The \
1247	digest over them held, so the bytes are the bytes that were written and \
1248	what is wrong is what they say.", ordinal, body.len();
1249	Decode, Input, Invalid)),
1250	}
1251	if out.len() > PACKED_MAX {
1252	return Err(err!(
1253	"The packed run at record {} inflates past {} bytes, which is the most a \
1254	run may come to. A run this crate writes is a megabyte, so this is not one \
1255	of them, and it is refused rather than allocated for.",
1256	ordinal, PACKED_MAX;
1257	Decode, Input, Excessive));
1258	}
1259	if out.is_empty() {
1260	return Err(err!(
1261	"The packed run at record {} inflates to nothing. A run carries the \
1262	records it packed, and an empty one cannot be told from a damaged one.",
1263	ordinal;
1264	Decode, Input, Missing));
1265	}
1266	Ok(out)
1267	}
1268
1269	/// A header's parents, named the way a reader names them.
1270	fn said_parents(head: &Header) -> String {
1271	let said: Vec<String> = head.parents().iter().map(\|p\| fmt!("{}", p)).collect();
1272	if said.is_empty() {
1273	fmt!("nothing, as a root")
1274	} else {
1275	said.join(", ")
1276	}
1277	}
1278
1279
1280	/// `None` means the bytes so far are a prefix of a varint and more are needed;
1281	/// an error means they are not a varint at all, whatever follows.
1282	fn try_varint(buf: &[u8])
1283	-> Outcome<Option<(u64, usize)>>
1284	{
1285	let ended = buf.iter().take(VARINT_MAX_LEN).any(\|b\| *b & 0x80 == 0);
1286	if !ended && buf.len() < VARINT_MAX_LEN {
1287	return Ok(None);
1288	}
1289	let (n, used) = res!(varint_decode(buf));
1290	Ok(Some((n, used)))
1291	}
1292
1293
1294	/// Writes a whole segment in one go.
1295	pub fn encode<H: Hasher, const S: usize>(
1296	head: &Head,
1297	entries: &[Entry],
1298	hasher: H,
1299	salt: [u8; S],
1300	)
1301	-> Outcome<Vec<u8>>
1302	{
1303	let mut writer: Writer<H, S> = Writer::new(head, hasher, salt);
1304	res!(writer.extend(entries));
1305	Ok(writer.finish())
1306	}
1307
1308	/// Reads a whole segment held in memory.
1309	pub fn decode<H: Hasher, const S: usize>(bytes: &[u8], hasher: H, salt: [u8; S])
1310	-> Outcome<(Head, Vec<Entry>)>
1311	{
1312	let mut reader: Reader<H, S> = Reader::new(hasher, salt);
1313	reader.feed(bytes);
1314	reader.end();
1315	let mut entries = Vec::new();
1316	while let Some(entry) = res!(reader.next_entry()) {
1317	entries.push(entry);
1318	}
1319	let head = match reader.head() {
1320	Some(h) => *h,
1321	None => return Err(err!(
1322	"A segment of {} bytes carries no header.", bytes.len();
1323	Decode, Input, Missing)),
1324	};
1325	Ok((head, entries))
1326	}
1327
1328
1329	#[cfg(test)]
1330	mod tests {
1331	use super::*;
1332
1333	use crate::id::{
1334	Anchor,
1335	ContentId,
1336	ContentRange,
1337	};
1338	use crate::op::{
1339	Header,
1340	Mode,
1341	Op,
1342	};
1343	use crate::op::tests::samples;
1344	use crate::test_support::{
1345	Fold,
1346	StubSigner,
1347	};
1348
1349	use oxedyne_fe2o3_iop_crypto::{
1350	InNamex,
1351	NamexId,
1352	keys::KeyManager,
1353	};
1354
1355	fn oid(replica: u64, counter: u64) -> OpId {
1356	OpId::new(ReplicaId::new(replica), counter)
1357	}
1358
1359	/// A stand-in cipher: the input under a keystream folded from the key, with
1360	/// four bytes of tag after it.
1361	///
1362	/// It is not cryptography and is offered as none. What these tests need of a
1363	/// cipher is three things: that the plaintext cannot be found in the output by
1364	/// searching for it, that only the same key gets it back, and that a wrong key
1365	/// fails rather than returning rubbish. An Ore repository veils under
1366	/// AES-256-GCM from `oxedyne_fe2o3_crypto`, which is tested where it is
1367	/// implemented; what is tested here is that this module puts the right bytes
1368	/// in front of a cipher and does the right thing with what comes back.
1369	#[derive(Clone, Debug, Default)]
1370	struct StubCipher {
1371	/// The shared key.
1372	key: Vec<u8>,
1373	}
1374
1375	impl StubCipher {
1376	fn with_seed(seed: u8) -> Self {
1377	Self { key: vec![seed; 16] }
1378	}
1379
1380	/// The fold both the keystream and the tag are drawn from.
1381	fn fold(key: &[u8], extra: &[u8]) -> u64 {
1382	let mut acc: u64 = 0xcbf2_9ce4_8422_2325;
1383	for b in key.iter().chain(extra.iter()) {
1384	acc ^= *b as u64;
1385	acc = acc.wrapping_mul(0x0000_0100_0000_01b3);
1386	}
1387	acc
1388	}
1389
1390	fn stream(&self, data: &[u8]) -> Vec<u8> {
1391	let mut acc = Self::fold(&self.key, &[]);
1392	data.iter()
1393	.map(\|b\| {
1394	acc = acc
1395	.wrapping_mul(6_364_136_223_846_793_005)
1396	.wrapping_add(1_442_695_040_888_963_407);
1397	b ^ (acc >> 33) as u8
1398	})
1399	.collect()
1400	}
1401	}
1402
1403	impl InNamex for StubCipher {
1404	fn name_id(&self) -> Outcome<NamexId> {
1405	Ok(NamexId::default())
1406	}
1407	}
1408
1409	impl KeyManager for StubCipher {
1410	fn clone_with_keys(&self, _pk: Option<&[u8]>, sk: Option<&[u8]>)
1411	-> Outcome<Self>
1412	{
1413	Ok(Self {
1414	key: match sk {
1415	Some(b) => b.to_vec(),
1416	None => Vec::new(),
1417	},
1418	})
1419	}
1420
1421	fn get_public_key(&self) -> Outcome<Option<&[u8]>> { Ok(None) }
1422
1423	fn get_secret_key(&self) -> Outcome<Option<&[u8]>> { Ok(Some(&self.key)) }
1424
1425	fn set_public_key(self, _pk: Option<&[u8]>) -> Outcome<Self> { Ok(self) }
1426
1427	fn set_secret_key(mut self, sk: Option<&[u8]>) -> Outcome<Self> {
1428	self.key = match sk {
1429	Some(b) => b.to_vec(),
1430	None => Vec::new(),
1431	};
1432	Ok(self)
1433	}
1434	}
1435
1436	impl Encrypter for StubCipher {
1437	fn encrypt(&self, data: &[u8])
1438	-> Outcome<Vec<u8>>
1439	{
1440	let mut out = self.stream(data);
1441	out.extend_from_slice(&Self::fold(&self.key, data).to_be_bytes()[..4]);
1442	Ok(out)
1443	}
1444
1445	fn decrypt(&self, data: &[u8])
1446	-> Outcome<Vec<u8>>
1447	{
1448	if data.len() < 4 {
1449	return Err(err!(
1450	"A body of {} bytes is shorter than the tag.", data.len();
1451	Decode, Input, Missing));
1452	}
1453	let cut = data.len() - 4;
1454	let plain = self.stream(&data[..cut]);
1455	if Self::fold(&self.key, &plain).to_be_bytes()[..4] != data[cut..] {
1456	return Err(err!(
1457	"The tag does not check out under this key."; Invalid, Input, Decrypt));
1458	}
1459	Ok(plain)
1460	}
1461
1462	fn is_identity(&self) -> bool { false }
1463	}
1464
1465	/// A handful of records spanning the vocabulary, with roots and merges among
1466	/// their headers.
1467	fn records() -> Outcome<Vec<Record>> {
1468	Ok(vec![
1469	Record::root(oid(1, 1), Op::FileCreate { path: b"notes.md".to_vec() }),
1470	Record::new(
1471	res!(Header::new(oid(1, 2), vec![oid(1, 1)])),
1472	Op::Splice {
1473	left: Some(Anchor::origin(oid(1, 1))),
1474	right: None,
1475	remove: Vec::new(),
1476	insert: b"the quick brown fox".to_vec().into(),
1477	},
1478	),
1479	Record::new(
1480	res!(Header::new(oid(2, 3), vec![oid(1, 2)])),
1481	Op::Move {
1482	src: vec![res!(ContentRange::new(oid(1, 2), 4, 9))],
1483	left: Some(Anchor::after(ContentId::new(oid(1, 2), 18))),
1484	right: None,
1485	},
1486	),
1487	Record::new(
1488	res!(Header::new(oid(3, 9), vec![oid(1, 2), oid(2, 3)])),
1489	Op::Splice {
1490	left: Some(Anchor::after(ContentId::new(oid(1, 2), 0))),
1491	right: Some(Anchor::before(ContentId::new(oid(1, 2), 1))),
1492	remove: vec![res!(ContentRange::new(oid(1, 2), 10, 15))],
1493	insert: vec![0x2a; 900].into(), // beyond a single byte length
1494	},
1495	),
1496	Record::new(
1497	res!(Header::new(oid(3, 10), vec![oid(3, 9)])),
1498	Op::FileRename { file: oid(1, 1), path: vec![0xff, 0x2f, 0x00] },
1499	),
1500	Record::new(
1501	res!(Header::new(oid(3, 11), vec![oid(3, 10)])),
1502	Op::FileMode { file: oid(1, 1), mode: Mode::Executable },
1503	),
1504	Record::new(
1505	res!(Header::new(oid(3, 12), vec![oid(3, 11)])),
1506	Op::FileDelete { file: oid(1, 1) },
1507	),
1508	Record::new(
1509	res!(Header::new(oid(3, 13), vec![oid(3, 12)])),
1510	Op::Mark { name: fmt!("release-caf\u{e9}"), body: None, time: None },
1511	),
1512	Record::new(
1513	res!(Header::new(oid(4, 14), vec![oid(3, 13)])),
1514	Op::Note {
1515	on: vec![res!(ContentRange::new(oid(1, 2), 4, 9))],
1516	text: b"the fox is doing the work here".to_vec(),
1517	},
1518	),
1519	])
1520	}
1521
1522	fn bare() -> Outcome<Vec<Entry>> {
1523	Ok(res!(records()).into_iter().map(Entry::Bare).collect())
1524	}
1525
1526	/// Those records sealed under a stand-in signer, and the signer.
1527	fn sealed()
1528	-> Outcome<(Vec<Entry>, StubSigner)>
1529	{
1530	let s = StubSigner::with_seed(19);
1531	let mut out = Vec::new();
1532	for rec in res!(records()) {
1533	out.push(Entry::Sealed(res!(Envelope::seal_record(&s, &rec))));
1534	}
1535	Ok((out, s))
1536	}
1537
1538	#[test]
1539	fn bare_records_round_trip() -> Outcome<()> {
1540	let entries = res!(bare());
1541	let head = Head::new(Some(ReplicaId::new(7)));
1542	let bytes = res!(encode(&head, &entries, Fold, [0u8; 0]));
1543	let (got_head, got) = res!(decode(&bytes, Fold, [0u8; 0]));
1544	assert_eq!(got_head, head);
1545	assert_eq!(got, entries);
1546	Ok(())
1547	}
1548
1549	/// Reads `want` entries from the front of a segment and says how many bytes
1550	/// they took, which is the only place a byte offset into a segment may come
1551	/// from.
1552	fn up_to(bytes: &[u8], want: usize)
1553	-> Outcome<(Head, Vec<Entry>, Vec<u8>, usize)>
1554	{
1555	// The header is read on the way to the first record, so a caller that
1556	// wants none of them has to read it for itself.
1557	if want == 0 {
1558	let (head, used) = res!(res!(Head::decode(bytes)).ok_or_else(\|\| err!(
1559	"The segment yielded no header."; Bug, Missing)));
1560	return Ok((head, Vec::new(), Vec::new(), used));
1561	}
1562	let mut reader: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1563	reader.feed(bytes);
1564	let mut got = Vec::new();
1565	while got.len() < want {
1566	match res!(reader.next_entry()) {
1567	Some(entry) => got.push(entry),
1568	None => break,
1569	}
1570	}
1571	let head = res!(reader.head().ok_or_else(\|\| err!(
1572	"The segment yielded no header."; Bug, Missing)));
1573	let at = bytes.len() - reader.remaining().len();
1574	Ok((*head, got, reader.take_digests(), at))
1575	}
1576
1577	#[test]
1578	fn a_reader_taken_up_part_way_yields_what_a_whole_read_yields() -> Outcome<()> {
1579	let entries = res!(bare());
1580	let head = Head::new(Some(ReplicaId::new(7)));
1581	let bytes = res!(encode(&head, &entries, Fold, [0u8; 0]));
1582	let (_, whole, all_digests, _) = res!(up_to(&bytes, entries.len() + 1));
1583	assert_eq!(whole, entries);
1584	// Every boundary, so that no one lucky stopping place carries the test.
1585	for stopped in 0..entries.len() {
1586	let (got_head, first, first_digests, at) = res!(up_to(&bytes, stopped));
1587	assert_eq!(got_head, head);
1588	let mut reader: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1589	reader.take_up(got_head, stopped);
1590	reader.feed(&bytes[at..]);
1591	reader.end();
1592	let mut rest = Vec::new();
1593	while let Some(entry) = res!(reader.next_entry()) {
1594	rest.push(entry);
1595	}
1596	let mut joined = first;
1597	joined.extend(rest);
1598	assert_eq!(joined, entries,
1599	"a read stopped after {} entries and taken up again lost or changed \
1600	something", stopped);
1601	let mut digests = first_digests;
1602	digests.extend(reader.take_digests());
1603	assert_eq!(digests, all_digests,
1604	"the digests of a read stopped after {} entries and taken up again are \
1605	not the digests of one read", stopped);
1606	assert_eq!(reader.count(), entries.len(),
1607	"a reader taken up at {} did not end at the count the file holds", stopped);
1608	}
1609	Ok(())
1610	}
1611
1612	#[test]
1613	fn a_reader_taken_up_names_a_damaged_record_by_its_place_in_the_file() -> Outcome<()> {
1614	let entries = res!(bare());
1615	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
1616	const STOPPED: usize = 4;
1617	let (head, _, _, at) = res!(up_to(&bytes, STOPPED));
1618	let mut damaged = bytes.to_vec();
1619	// Into the body of the record that begins there: past its kind byte and
1620	// the varint that gives its length.
1621	damaged[at + 3] ^= 0xff;
1622	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
1623	reader.take_up(head, STOPPED);
1624	reader.feed(&damaged[at..]);
1625	reader.end();
1626	match reader.next_entry() {
1627	Ok(_) => Err(err!(
1628	"A record whose body was altered was read as though it were sound.";
1629	Test, Invalid)),
1630	Err(e) => {
1631	let said = fmt!("{}", e);
1632	assert!(said.contains(&fmt!("Record {} of the segment", STOPPED)),
1633	"the message names the record by its place in the read rather than \
1634	in the file: {}", said);
1635	Ok(())
1636	},
1637	}
1638	}
1639
1640	#[test]
1641	fn sealed_records_round_trip_and_still_verify() -> Outcome<()> {
1642	let (entries, signer) = res!(sealed());
1643	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
1644	let (head, got) = res!(decode(&bytes, Fold, [0u8; 0]));
1645	assert_eq!(head.replica, None);
1646	assert_eq!(got, entries);
1647	for entry in &got {
1648	match entry {
1649	Entry::Sealed(e) => {
1650	assert!(res!(e.verify(&signer)));
1651	assert!(res!(e.open_record(&signer)).parents().len() <= 2);
1652	},
1653	other => return Err(err!(
1654	"Expected a sealed envelope, got a {}.", other.name();
1655	Test, Mismatch)),
1656	}
1657	}
1658	Ok(())
1659	}
1660
1661	#[test]
1662	fn both_forms_mix_in_one_segment() -> Outcome<()> {
1663	let (mut entries, _) = res!(sealed());
1664	entries.truncate(2);
1665	entries.extend(res!(bare()));
1666	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
1667	let (_, got) = res!(decode(&bytes, Fold, [0u8; 0]));
1668	assert_eq!(got, entries);
1669	assert_eq!(got[0].kind(), KIND_SEALED);
1670	assert_eq!(got[2].kind(), KIND_BARE);
1671	Ok(())
1672	}
1673
1674	/// Both forms survive the tagged daticle round trip, which is what a carrier
1675	/// that is itself a daticle uses, and a tag that is neither is refused.
1676	#[test]
1677	fn entries_round_trip_as_daticles() -> Outcome<()> {
1678	let (mut entries, _) = res!(sealed());
1679	entries.extend(res!(bare()));
1680	for entry in &entries {
1681	assert_eq!(&res!(Entry::from_dat(&entry.to_dat())), entry);
1682	}
1683	let odd = Dat::List(vec![Dat::U8(9), Dat::List(Vec::new())]);
1684	assert!(Entry::from_dat(&odd).is_err());
1685	assert!(Entry::from_dat(&Dat::List(Vec::new())).is_err());
1686	Ok(())
1687	}
1688
1689	#[test]
1690	fn an_empty_segment_is_just_a_header() -> Outcome<()> {
1691	let head = Head::new(Some(ReplicaId::new(0)));
1692	let bytes = res!(encode(&head, &[], Fold, [0u8; 0]));
1693	assert_eq!(bytes, head.encode());
1694	let (got_head, got) = res!(decode(&bytes, Fold, [0u8; 0]));
1695	assert_eq!(got_head, head);
1696	assert!(got.is_empty());
1697	Ok(())
1698	}
1699
1700	#[test]
1701	fn a_byte_at_a_time_reads_the_same() -> Outcome<()> {
1702	let entries = res!(bare());
1703	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
1704	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
1705	let mut got: Vec<Entry> = Vec::new();
1706	for b in &bytes {
1707	reader.feed(&[*b]);
1708	while let Some(entry) = res!(reader.next_entry()) {
1709	got.push(entry);
1710	}
1711	}
1712	reader.end();
1713	while let Some(entry) = res!(reader.next_entry()) {
1714	got.push(entry);
1715	}
1716	assert_eq!(got, entries);
1717	assert!(reader.is_exhausted());
1718	assert_eq!(reader.count(), entries.len());
1719	Ok(())
1720	}
1721
1722	/// A tallying reader hands back the digests it checked, in order, and hands
1723	/// back exactly those.
1724	///
1725	/// The oracle is the segment itself: every digest the reader reports must be
1726	/// found in the encoded bytes, and at a higher offset than the one before it.
1727	/// That is independent of how the reader computed them, which is what makes
1728	/// it worth asserting -- a tally built by hashing something else, or built in
1729	/// the wrong order, or one digest short, fails it.
1730	#[test]
1731	fn a_tallying_reader_reports_the_digests_it_checked() -> Outcome<()> {
1732	let entries = res!(bare());
1733	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
1734
1735	// Taken once at the end, so that the take holds every digest at once and
1736	// the order they come back in is a thing this can be wrong about. An
1737	// earlier version took after every record, and a take of one digest is in
1738	// order whatever the reader does with it.
1739	let mut whole: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1740	whole.feed(&bytes);
1741	whole.end();
1742	while let Some(_) = res!(whole.next_entry()) {}
1743	let tally = whole.take_digests();
1744	assert_eq!(tally.len(), entries.len() * 8, "one eight byte digest per record");
1745	assert!(whole.take_digests().is_empty(), "and a second take has nothing left");
1746	assert!(entries.len() > 2, "the fixture must hold enough records to be out of order");
1747
1748	let mut at = 0usize;
1749	for (i, digest) in tally.chunks(8).enumerate() {
1750	let found = match bytes[at..].windows(8).position(\|w\| w == digest) {
1751	Some(p) => at + p,
1752	None => return Err(err!(
1753	"The digest reported for record {} is not in the segment after \
1754	offset {}.", i, at; Test, Mismatch)),
1755	};
1756	at = found + 1;
1757	}
1758
1759	// The chunking the bytes arrive in changes nothing, as it changes nothing
1760	// about the records, and neither does draining the tally as it fills,
1761	// which is what a reader working a batch at a time does.
1762	let mut dribbled: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1763	let mut slow = Vec::new();
1764	for b in &bytes {
1765	dribbled.feed(&[*b]);
1766	while let Some(_) = res!(dribbled.next_entry()) {
1767	slow.extend_from_slice(&dribbled.take_digests());
1768	}
1769	}
1770	dribbled.end();
1771	while let Some(_) = res!(dribbled.next_entry()) {
1772	slow.extend_from_slice(&dribbled.take_digests());
1773	}
1774	slow.extend_from_slice(&dribbled.take_digests());
1775	assert_eq!(slow, tally, "a byte at a time tallies what a mouthful tallies");
1776
1777	// A reader nobody asked reports nothing, and reads the same records.
1778	let mut plain: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
1779	plain.feed(&bytes);
1780	plain.end();
1781	let mut got = Vec::new();
1782	while let Some(entry) = res!(plain.next_entry()) {
1783	got.push(entry);
1784	assert!(plain.take_digests().is_empty(), "and keeps nothing on the way");
1785	}
1786	assert_eq!(got, entries);
1787	Ok(())
1788	}
1789
1790	/// The invariant packing rests on: a packed segment yields the same
1791	/// records, with the same digests, in the same order, as the plain segment it
1792	/// was made from.
1793	///
1794	/// A fold over those digests is what `ore repack` compares two stores by, so
1795	/// if this were not exact a compressed store and an uncompressed one carrying
1796	/// one history would disagree about their own shape, and packing would stop
1797	/// being revocable. The digests are compared as well as the entries, because
1798	/// the entries could agree while the framing they were checked against did
1799	/// not.
1800	#[test]
1801	fn a_packed_segment_yields_what_the_plain_one_yields() -> Outcome<()> {
1802	let entries = res!(bare());
1803	assert!(entries.len() > 2, "the fixture holds enough records to be a run");
1804	let head = Head::new(Some(ReplicaId::new(4)));
1805
1806	let plain = res!(encode(&head, &entries, Fold, [0u8; 0]));
1807	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
1808	res!(writer.push_packed(&entries));
1809	let packed = writer.finish();
1810	assert_ne!(plain, packed, "the two are not the same bytes");
1811
1812	let read = \|bytes: &[u8]\| -> Outcome<(Vec<Entry>, Vec<u8>, usize)> {
1813	let mut reader: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1814	reader.feed(bytes);
1815	reader.end();
1816	let mut got = Vec::new();
1817	while let Some(entry) = res!(reader.next_entry()) {
1818	got.push(entry);
1819	}
1820	assert!(reader.is_exhausted(), "every byte became a record");
1821	let tally = reader.take_digests();
1822	Ok((got, tally, reader.count()))
1823	};
1824	let (plain_entries, plain_tally, plain_count) = res!(read(&plain));
1825	let (packed_entries, packed_tally, packed_count) = res!(read(&packed));
1826
1827	assert_eq!(plain_entries, entries, "the plain segment reads back");
1828	assert_eq!(packed_entries, entries, "and so does the packed one");
1829	assert_eq!(packed_count, plain_count, "the same number of records");
1830	assert_eq!(packed_count, entries.len(), "which is the number that went in");
1831	assert_eq!(packed_tally, plain_tally,
1832	"and the same digests in the same order, which is what a fold over a log \
1833	names and what makes packing revocable");
1834	assert!(!packed_tally.is_empty(), "the fixture really tallied something");
1835	Ok(())
1836	}
1837
1838	/// A run's own digest is over the compressed bytes and is NOT among the
1839	/// digests a reader tallies.
1840	///
1841	/// Stated separately because it is the part that would be easy to get right
1842	/// by accident and wrong on the next change: one packed record yields several
1843	/// records, and what a fold wants is the several.
1844	#[test]
1845	fn the_runs_own_digest_is_not_tallied() -> Outcome<()> {
1846	let entries = res!(bare());
1847	let head = Head::new(None);
1848	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
1849	res!(writer.push_packed(&entries));
1850	let packed = writer.finish();
1851
1852	let mut reader: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1853	reader.feed(&packed);
1854	reader.end();
1855	while res!(reader.next_entry()).is_some() {}
1856	let tally = reader.take_digests();
1857	assert_eq!(tally.len(), entries.len() * 8,
1858	"one eight byte digest per RECORD, not one for the run");
1859	Ok(())
1860	}
1861
1862	/// The framing around a packed run is fixed, and its payload is not frozen.
1863	///
1864	/// There is no golden byte array here on purpose. A packed run's payload is
1865	/// what a compressor made of the records, so freezing it would freeze a
1866	/// dependency version and call it a format: the first `cargo update` that
1867	/// moved `miniz_oxide` would redden it, with nothing about Ore having
1868	/// changed. What a reader elsewhere must agree about is the framing, and that
1869	/// is what is asserted -- the magic, the declared version, the kind byte, and
1870	/// that the length and the digest that follow describe what is there.
1871	#[test]
1872	fn the_packed_framing_is_fixed() -> Outcome<()> {
1873	let entries = res!(bare());
1874	let head = Head::new(None);
1875	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
1876	res!(writer.push_packed(&entries));
1877	let bytes = writer.finish();
1878
1879	assert_eq!(&bytes[..MAGIC.len()], &MAGIC[..], "a segment begins with the magic");
1880	assert_eq!(bytes[6], VERSION, "the version sits where it always has");
1881	assert_eq!(bytes[7], 0, "no replica hint follows");
1882	assert_eq!(bytes[8], KIND_PACKED, "and the record says it is a run");
1883
1884	// The length, the payload and the digest, read the way a reader reads them.
1885	let (len, used) = res!(varint_decode(&bytes[9..]));
1886	let at = 9 + used;
1887	let body = &bytes[at..at + len as usize];
1888	let (dlen, used) = res!(varint_decode(&bytes[at + len as usize..]));
1889	let dat = at + len as usize + used;
1890	assert_eq!(bytes.len(), dat + dlen as usize, "and nothing after the digest");
1891	let want = Fold.hash(&[&[KIND_PACKED], body], [0u8; 0]).as_vec();
1892	assert_eq!(&bytes[dat..], &want[..],
1893	"the digest is over the compressed bytes, which is what catches damage \
1894	before anything is inflated");
1895	Ok(())
1896	}
1897
1898	/// A run inside a run is refused.
1899	#[test]
1900	fn a_run_inside_a_run_is_refused() -> Outcome<()> {
1901	let entries = res!(bare());
1902	let head = Head::new(None);
1903	// A run of records, packed, and then that whole framing packed again as if
1904	// it were a run of its own.
1905	let mut inner: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
1906	res!(inner.push_packed(&entries));
1907	let once = inner.finish();
1908	let framing = &once[Head::new(None).encode().len()..];
1909
1910	let mut outer: Vec<u8> = Head::new(None).encode();
1911	let body = res!(deflate(framing));
1912	let digest = Fold.hash(&[&[KIND_PACKED], &body], [0u8; 0]).as_vec();
1913	outer.push(KIND_PACKED);
1914	varint_encode(body.len() as u64, &mut outer);
1915	outer.extend_from_slice(&body);
1916	varint_encode(digest.len() as u64, &mut outer);
1917	outer.extend_from_slice(&digest);
1918
1919	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
1920	reader.feed(&outer);
1921	reader.end();
1922	let said = match reader.next_entry() {
1923	Ok(_) => return Err(err!("A run inside a run was read."; Test, Invalid)),
1924	Err(e) => fmt!("{}", e.plain()),
1925	};
1926	assert!(said.contains("itself packed"), "and says so: {}", said);
1927	Ok(())
1928	}
1929
1930	/// A damaged run is named by its digest, before a byte of it is inflated.
1931	#[test]
1932	fn a_damaged_run_is_named_and_never_inflated() -> Outcome<()> {
1933	let entries = res!(bare());
1934	let head = Head::new(None);
1935	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
1936	res!(writer.push_packed(&entries));
1937	let good = writer.finish();
1938
1939	// Every single-byte change to the compressed payload, which is where damage
1940	// lands: each is caught, and none of them reaches the decompressor.
1941	let at = 9 + res!(varint_decode(&good[9..])).1;
1942	let len = res!(varint_decode(&good[9..])).0 as usize;
1943	assert!(len > 4, "there is a payload to damage");
1944	for i in [at, at + 1, at + len / 2, at + len - 1] {
1945	let mut bad = good.clone();
1946	bad[i] ^= 0x01;
1947	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
1948	reader.feed(&bad);
1949	reader.end();
1950	let said = match reader.next_entry() {
1951	Ok(_) => return Err(err!(
1952	"A run damaged at byte {} was read.", i; Test, Invalid)),
1953	Err(e) => fmt!("{}", e.plain()),
1954	};
1955	assert!(said.contains("fails its integrity check"),
1956	"damage at {} is caught by the digest, not by the decompressor: {}",
1957	i, said);
1958	assert!(said.contains("a run of packed operations"),
1959	"and the message says what the record was: {}", said);
1960	}
1961	Ok(())
1962	}
1963
1964	/// A vouched read yields exactly what a checked read yields, for every shape
1965	/// of segment there is.
1966	///
1967	/// This is the whole of what the gate may change: which bodies get hashed.
1968	/// The records that come out, the order they come out in, the header and the
1969	/// tally a fold is built from must all be the same bytes, because a caller
1970	/// switching modes is not asking for a different history. The tally matters
1971	/// most: a vouched read hands back the digest it found rather than one it
1972	/// computed, and if those two ever differed on sound bytes then every verdict
1973	/// ever filed would miss.
1974	#[test]
1975	fn a_vouched_read_yields_what_a_checked_read_yields() -> Outcome<()> {
1976	let entries = res!(bare());
1977	let (signed, _) = res!(sealed());
1978	let head = Head::new(Some(ReplicaId::new(7)));
1979
1980	let mut mixed: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
1981	res!(mixed.push(&entries[0]));
1982	res!(mixed.push_packed(&entries[1..]));
1983	res!(mixed.push(&signed[0]));
1984	let shapes = [
1985	("plain bare", res!(encode(&head, &entries, Fold, [0u8; 0]))),
1986	("plain sealed", res!(encode(&head, &signed, Fold, [0u8; 0]))),
1987	("packed and plain together", mixed.finish()),
1988	];
1989
1990	for (shape, bytes) in &shapes {
1991	let mut want: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
1992	let mut got: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0])
1993	.integrity(Integrity::Vouched);
1994	let (mut want_out, mut got_out) = (Vec::new(), Vec::new());
1995	want.feed(bytes);
1996	want.end();
1997	got.feed(bytes);
1998	got.end();
1999	while let Some(entry) = res!(want.next_entry()) {
2000	want_out.push(entry);
2001	}
2002	while let Some(entry) = res!(got.next_entry()) {
2003	got_out.push(entry);
2004	}
2005	assert!(!want_out.is_empty(), "the {} fixture holds records", shape);
2006	assert_eq!(got_out, want_out, "a vouched read of a {} segment", shape);
2007	assert_eq!(got.head(), want.head(), "and reads the same header, {}", shape);
2008	assert_eq!(got.count(), want.count(), "and the same count, {}", shape);
2009	let tally = want.take_digests();
2010	assert_eq!(tally.len(), want_out.len() * 8, "one digest per record, {}", shape);
2011	assert_eq!(got.take_digests(), tally,
2012	"and the same tally, so a fold over a vouched read is the fold a \
2013	verdict was filed under, {}", shape);
2014	}
2015	Ok(())
2016	}
2017
2018	/// The trade, written down. A body byte that flips under a vouched read
2019	/// is handed over as though nothing happened, and a fold cannot see it
2020	/// either.
2021	///
2022	/// The same damage under a checked read is refused by name. Both halves are
2023	/// asserted here because the pair is the point: this is not a test that the
2024	/// gate works, it is a test of what the gate costs, and the cost is what
2025	/// [`crate::segment::Integrity`] tells a caller to go and cover somewhere
2026	/// else.
2027	#[test]
2028	fn a_vouched_read_lets_a_flipped_body_byte_through() -> Outcome<()> {
2029	let entries = res!(bare());
2030	let head = Head::new(None);
2031	let good = res!(encode(&head, &entries, Fold, [0u8; 0]));
2032
2033	// One letter of a note's text, which is bit rot as it actually reads: a
2034	// byte that keeps its record the same length and the same shape, so
2035	// nothing but the digest over it could ever have noticed.
2036	let at = res!(good.windows(3).position(\|w\| w == b"fox").ok_or_else(\|\| err!(
2037	"The fixture no longer carries the text this test damages."; Test, Missing)));
2038	let mut bad = good.clone();
2039	bad[at] ^= 0x20;
2040	assert_eq!(bad.len(), good.len(), "the damage moved nothing");
2041	assert_ne!(bad, good, "and really is damage");
2042
2043	let mut checked: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
2044	checked.feed(&bad);
2045	checked.end();
2046	let said = loop {
2047	match checked.next_entry() {
2048	Ok(Some(_)) => (),
2049	Ok(None) => return Err(err!(
2050	"A checked read took a segment whose body bytes had been changed.";
2051	Test, Invalid)),
2052	Err(e) => break fmt!("{}", e.plain()),
2053	}
2054	};
2055	assert!(said.contains("fails its integrity check"),
2056	"a checked read still says what is wrong: {}", said);
2057
2058	let mut vouched: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0])
2059	.integrity(Integrity::Vouched);
2060	vouched.feed(&bad);
2061	vouched.end();
2062	let mut got = Vec::new();
2063	while let Some(entry) = res!(vouched.next_entry()) {
2064	got.push(entry);
2065	}
2066	assert_eq!(got.len(), entries.len(),
2067	"a vouched read hands over every record of a damaged segment");
2068	assert_ne!(got, entries,
2069	"and what it hands over is not what was written");
2070
2071	// And the fold is blind to it, which is the part that decides where the
2072	// check has to go instead: the digests are read out of the file, so the
2073	// same file with a changed body folds to what it folded to before.
2074	let mut sound: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0])
2075	.integrity(Integrity::Vouched);
2076	sound.feed(&good);
2077	sound.end();
2078	while let Some(_) = res!(sound.next_entry()) {}
2079	assert_eq!(vouched.take_digests(), sound.take_digests(),
2080	"a damaged body folds to what the sound one folded to, so nothing \
2081	downstream of the tally can catch this either");
2082	Ok(())
2083	}
2084
2085	/// A vouched read reaches the decompressor with the frame unexamined, and the
2086	/// bound still refuses a run that would not stop.
2087	///
2088	/// [`Integrity::Vouched`] gives up the digest that used to stand in front of
2089	/// [`inflate`], so the only thing between a hostile frame and the allocator
2090	/// is [`PACKED_MAX`]. That was true before and is load bearing now.
2091	#[test]
2092	fn a_vouched_read_still_refuses_a_run_that_would_not_stop() -> Outcome<()> {
2093	let big = vec![0u8; PACKED_MAX + 1024];
2094	let body = res!(deflate(&big));
2095	let mut bytes = Head::new(None).encode();
2096	bytes.push(KIND_PACKED);
2097	varint_encode(body.len() as u64, &mut bytes);
2098	bytes.extend_from_slice(&body);
2099	// A digest that is not the frame's, so that nothing here could be passing
2100	// because the frame happened to check out.
2101	varint_encode(8, &mut bytes);
2102	bytes.extend_from_slice(&[0u8; 8]);
2103
2104	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0])
2105	.integrity(Integrity::Vouched);
2106	reader.feed(&bytes);
2107	reader.end();
2108	let said = match reader.next_entry() {
2109	Ok(_) => return Err(err!(
2110	"A vouched read inflated a run past the bound."; Test, Invalid)),
2111	Err(e) => fmt!("{}", e.plain()),
2112	};
2113	assert!(said.contains("inflates past"), "and says why: {}", said);
2114	Ok(())
2115	}
2116
2117	/// A run that would not stop inflating is refused rather than allocated for.
2118	///
2119	/// The digest holds, so this is not damage: it is a frame somebody wrote to be
2120	/// obeyed. What refuses it is the bound and nothing else, which is why the
2121	/// frame is built to be sound in every other respect.
2122	#[test]
2123	fn a_run_that_would_not_stop_is_refused() -> Outcome<()> {
2124	let big = vec![0u8; PACKED_MAX + 1024];
2125	let body = res!(deflate(&big));
2126	assert!(body.len() < 1 << 20, "the fixture really is a small frame: {}", body.len());
2127	let digest = Fold.hash(&[&[KIND_PACKED], &body], [0u8; 0]).as_vec();
2128	let mut bytes = Head::new(None).encode();
2129	bytes.push(KIND_PACKED);
2130	varint_encode(body.len() as u64, &mut bytes);
2131	bytes.extend_from_slice(&body);
2132	varint_encode(digest.len() as u64, &mut bytes);
2133	bytes.extend_from_slice(&digest);
2134
2135	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2136	reader.feed(&bytes);
2137	reader.end();
2138	let said = match reader.next_entry() {
2139	Ok(_) => return Err(err!("A run past the bound was inflated."; Test, Invalid)),
2140	Err(e) => fmt!("{}", e.plain()),
2141	};
2142	assert!(said.contains("inflates past"), "and says why: {}", said);
2143	Ok(())
2144	}
2145
2146	/// Packed and plain records sit side by side in one segment.
2147	///
2148	/// The kind is per record, so a segment is not one thing or the other. This is
2149	/// what lets a repack pack the sealed part of a log and leave the tail alone.
2150	#[test]
2151	fn a_segment_holds_packed_and_plain_together() -> Outcome<()> {
2152	let entries = res!(bare());
2153	let head = Head::new(None);
2154	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
2155	res!(writer.push(&entries[0]));
2156	res!(writer.push_packed(&entries[1..]));
2157	res!(writer.push(&entries[0]));
2158	let bytes = writer.finish();
2159
2160	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2161	reader.feed(&bytes);
2162	reader.end();
2163	let mut got = Vec::new();
2164	while let Some(entry) = res!(reader.next_entry()) {
2165	got.push(entry);
2166	}
2167	let mut want = vec![entries[0].clone()];
2168	want.extend_from_slice(&entries[1..]);
2169	want.push(entries[0].clone());
2170	assert_eq!(got, want, "in the order they were written");
2171	assert!(reader.is_exhausted());
2172	Ok(())
2173	}
2174
2175	/// A byte at a time reads a packed segment as a mouthful does.
2176	#[test]
2177	fn a_byte_at_a_time_reads_a_packed_segment_the_same() -> Outcome<()> {
2178	let entries = res!(bare());
2179	let head = Head::new(None);
2180	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
2181	res!(writer.push_packed(&entries));
2182	let bytes = writer.finish();
2183
2184	let mut reader: Reader<Fold, 0> = Reader::tallying(Fold, [0u8; 0]);
2185	let mut got: Vec<Entry> = Vec::new();
2186	let mut slow = Vec::new();
2187	for b in &bytes {
2188	reader.feed(&[*b]);
2189	while let Some(entry) = res!(reader.next_entry()) {
2190	got.push(entry);
2191	}
2192	slow.extend_from_slice(&reader.take_digests());
2193	}
2194	reader.end();
2195	while let Some(entry) = res!(reader.next_entry()) {
2196	got.push(entry);
2197	}
2198	slow.extend_from_slice(&reader.take_digests());
2199	assert_eq!(got, entries, "a run only becomes records once all of it has arrived");
2200	assert_eq!(slow.len(), entries.len() * 8);
2201	assert!(reader.is_exhausted());
2202	Ok(())
2203	}
2204
2205	/// An empty run is refused at both ends.
2206	#[test]
2207	fn an_empty_run_is_refused() -> Outcome<()> {
2208	let head = Head::new(None);
2209	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
2210	let said = match writer.push_packed(&[]) {
2211	Ok(()) => return Err(err!("An empty run was written."; Test, Invalid)),
2212	Err(e) => fmt!("{}", e.plain()),
2213	};
2214	assert!(said.contains("over no entries"), "and says so: {}", said);
2215
2216	// And one that inflates to nothing, which is what a reader could meet.
2217	let body = res!(deflate(&[]));
2218	let digest = Fold.hash(&[&[KIND_PACKED], &body], [0u8; 0]).as_vec();
2219	let mut bytes = Head::new(None).encode();
2220	bytes.push(KIND_PACKED);
2221	varint_encode(body.len() as u64, &mut bytes);
2222	bytes.extend_from_slice(&body);
2223	varint_encode(digest.len() as u64, &mut bytes);
2224	bytes.extend_from_slice(&digest);
2225	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2226	reader.feed(&bytes);
2227	reader.end();
2228	let said = match reader.next_entry() {
2229	Ok(_) => return Err(err!("An empty run was read."; Test, Invalid)),
2230	Err(e) => fmt!("{}", e.plain()),
2231	};
2232	assert!(said.contains("inflates to nothing"), "and says so: {}", said);
2233	Ok(())
2234	}
2235
2236	#[test]
2237	fn the_header_arrives_before_the_records() -> Outcome<()> {
2238	let head = Head::new(Some(ReplicaId::new(300)));
2239	let bytes = res!(encode(&head, &res!(bare()), Fold, [0u8; 0]));
2240	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2241	reader.feed(&bytes[..head.encode().len()]);
2242	assert!(res!(reader.next_entry()).is_none());
2243	assert_eq!(reader.head(), Some(&head));
2244	Ok(())
2245	}
2246
2247	/// Truncating a segment anywhere is a typed error or a request for more
2248	/// bytes, never a panic and never a half-read record.
2249	#[test]
2250	fn truncation_at_every_offset_is_clean() -> Outcome<()> {
2251	let entries = res!(bare());
2252	let bytes = res!(encode(&Head::new(Some(ReplicaId::new(2))), &entries, Fold, [0u8; 0]));
2253	for cut in 0..bytes.len() {
2254	// Declared complete: a partial record is an error.
2255	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2256	reader.feed(&bytes[..cut]);
2257	reader.end();
2258	let mut whole = 0usize;
2259	loop {
2260	match reader.next_entry() {
2261	Ok(Some(_)) => whole += 1,
2262	Ok(None) \| Err(_) => break,
2263	}
2264	}
2265	assert!(whole < entries.len(), "cut at {} yielded every record", cut);
2266	// Not yet declared complete: the reader asks for more rather than
2267	// failing, unless the bytes are already wrong.
2268	let mut open: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2269	open.feed(&bytes[..cut]);
2270	loop {
2271	match open.next_entry() {
2272	Ok(Some(_)) => {},
2273	Ok(None) => break,
2274	Err(e) => return Err(err!(e,
2275	"Cut at {} of {} failed before the segment was declared \
2276	complete.", cut, bytes.len(); Test)),
2277	}
2278	}
2279	}
2280	// The whole segment reads every record.
2281	let (_, got) = res!(decode(&bytes, Fold, [0u8; 0]));
2282	assert_eq!(got.len(), entries.len());
2283	Ok(())
2284	}
2285
2286	#[test]
2287	fn a_damaged_record_is_named() -> Outcome<()> {
2288	let entries = res!(bare());
2289	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
2290	// Damage a byte of the first record's file name, so that the record still
2291	// decodes and the error can say which operation is at risk.
2292	let at = match bytes.windows(5).position(\|w\| w == b"notes") {
2293	Some(i) => i,
2294	None => return Err(err!(
2295	"The segment does not contain the file name it was built with.";
2296	Test, Missing)),
2297	};
2298	let mut damaged = bytes.clone();
2299	damaged[at] ^= 0x20;
2300	let e = match decode(&damaged, Fold, [0u8; 0]) {
2301	Ok(_) => return Err(err!(
2302	"A damaged record was accepted."; Test, Mismatch)),
2303	Err(e) => e,
2304	};
2305	let msg = fmt!("{}", e);
2306	assert!(msg.contains("integrity check"), "message was {:?}", msg);
2307	assert!(msg.contains("Record 0"), "message was {:?}", msg);
2308	assert!(msg.contains("r1:1"), "message was {:?}", msg);
2309	Ok(())
2310	}
2311
2312	#[test]
2313	fn a_damaged_digest_is_caught() -> Outcome<()> {
2314	let entries = res!(bare());
2315	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
2316	let mut damaged = bytes.clone();
2317	let last = damaged.len() - 1;
2318	damaged[last] ^= 0xff;
2319	assert!(decode(&damaged, Fold, [0u8; 0]).is_err());
2320	Ok(())
2321	}
2322
2323	/// A kind byte flipped from bare to sealed is caught by the digest, which
2324	/// covers it.
2325	#[test]
2326	fn the_kind_byte_is_covered_by_the_digest() -> Outcome<()> {
2327	let entries = res!(bare());
2328	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
2329	let head_len = Head::new(None).encode().len();
2330	let mut damaged = bytes.clone();
2331	assert_eq!(damaged[head_len], KIND_BARE);
2332	damaged[head_len] = KIND_SEALED;
2333	assert!(decode(&damaged, Fold, [0u8; 0]).is_err());
2334	Ok(())
2335	}
2336
2337	/// Reading with the wrong hasher or the wrong salt is the same failure as
2338	/// reading damaged bytes, which is what makes the check the caller's to own.
2339	#[test]
2340	fn the_hasher_must_match() -> Outcome<()> {
2341	let entries = res!(bare());
2342	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
2343	assert!(decode(&bytes, Fold, [1u8; 4]).is_err(), "a different salt");
2344	assert!(decode(&bytes, (), [0u8; 0]).is_err(), "a different function");
2345	// Under the identity hasher the digest is the body itself, and that too
2346	// round trips.
2347	let identity = res!(encode(&Head::new(None), &entries, (), [0u8; 0]));
2348	let (_, got) = res!(decode(&identity, (), [0u8; 0]));
2349	assert_eq!(got, entries);
2350	assert!(identity.len() > bytes.len(), "the identity digest costs the body twice");
2351	Ok(())
2352	}
2353
2354	#[test]
2355	fn a_segment_that_is_not_one_is_refused() -> Outcome<()> {
2356	assert!(decode(b"not a segment at all", Fold, [0u8; 0]).is_err());
2357	assert!(decode(b"O", Fold, [0u8; 0]).is_err(), "a truncated header");
2358	assert!(decode(b"X", Fold, [0u8; 0]).is_err(), "a wrong first byte");
2359	// The right magic at an unknown version is refused, and says so.
2360	let mut wrong = MAGIC.to_vec();
2361	wrong.push(VERSION + 1);
2362	wrong.push(0);
2363	let e = match decode(&wrong, Fold, [0u8; 0]) {
2364	Ok(_) => return Err(err!("An unknown version was accepted."; Test)),
2365	Err(e) => e,
2366	};
2367	assert!(fmt!("{}", e).contains("version"), "message was {}", e);
2368	// A record tagged with none of the kinds is refused, and the refusal names
2369	// the ones there are. That is the mechanism by which a build made before a
2370	// form existed meets it: a reader knowing only the bare and sealed kinds
2371	// says so about a veiled record in exactly these words, which is the whole
2372	// of what a new entry form owes an old reader.
2373	let entries = res!(bare());
2374	let bytes = res!(encode(&Head::new(None), &entries, (), [0u8; 0]));
2375	let head_len = Head::new(None).encode().len();
2376	// Under the identity hasher a record's digest is its kind byte and its body
2377	// again, so the tag is put right in the digest as well. Without that the
2378	// integrity check refuses the record before the tag is ever looked at,
2379	// which is what this assertion was quietly testing instead.
2380	let mut odd = bytes.clone();
2381	odd[head_len] = 9;
2382	let (body_len, used) = res!(varint_decode(&odd[head_len + 1..]));
2383	let after_body = head_len + 1 + used + body_len as usize;
2384	let (digest_len, used) = res!(varint_decode(&odd[after_body..]));
2385	assert_eq!(digest_len, body_len + 1, "the identity digest is the kind and the body");
2386	odd[after_body + used] = 9;
2387	let e = match decode(&odd, (), [0u8; 0]) {
2388	Ok(_) => return Err(err!("A record tagged 9 was accepted."; Test)),
2389	Err(e) => e,
2390	};
2391	let msg = fmt!("{}", e);
2392	for named in ["bare", "sealed", "veiled"] {
2393	assert!(msg.contains(named),
2394	"the refusal does not name the {} form: {}", named, msg);
2395	}
2396	Ok(())
2397	}
2398
2399	/// A version 2 segment reads, a version 1 segment does not, and the refusal
2400	/// says which versions this reader knows.
2401	///
2402	/// This is the half of the version 3 event that costs a repository nothing:
2403	/// version 2 held codes 1 to 7 and version 3 holds 1 to 8, so every segment
2404	/// written before the bump means under the new reader exactly what it meant
2405	/// under the old one. Version 1 is a different matter -- its operations
2406	/// spelled a file as a path -- and stays refused.
2407	#[test]
2408	fn a_version_two_segment_still_reads() -> Outcome<()> {
2409	// Everything but the FileMode, which is the one operation version 2 has
2410	// no code for.
2411	let entries: Vec<Entry> = res!(bare())
2412	.into_iter()
2413	.filter(\|e\| !matches!(e.peek(), Ok(Record { op: Op::FileMode { .. }, .. })))
2414	.collect();
2415	let old = Head { version: VERSION_MIN, replica: Some(ReplicaId::new(7)) };
2416	let bytes = res!(encode(&old, &entries, Fold, [0u8; 0]));
2417	assert_eq!(bytes[MAGIC.len()], VERSION_MIN, "the segment declares version 2");
2418	let (head, got) = res!(decode(&bytes, Fold, [0u8; 0]));
2419	assert_eq!(head, old, "the header reads back at the version it was written");
2420	assert_eq!(got, entries, "and every record with it");
2421	// Version 1 is below what this reader knows, and the message says so.
2422	let mut ancient = bytes.clone();
2423	ancient[MAGIC.len()] = VERSION_MIN - 1;
2424	let e = match decode(&ancient, Fold, [0u8; 0]) {
2425	Ok(_) => return Err(err!("Version 1 was accepted."; Test)),
2426	Err(e) => e,
2427	};
2428	let msg = fmt!("{}", e);
2429	assert!(msg.contains(&fmt!("{}", VERSION_MIN)), "message was {}", msg);
2430	assert!(msg.contains(&fmt!("{}", VERSION)), "message was {}", msg);
2431	Ok(())
2432	}
2433
2434	/// A segment declaring an older version will not be given an operation that
2435	/// version has no code for.
2436	///
2437	/// Without this the subset claim would hold of the intention and not of the
2438	/// bytes: appending a FileMode to a version 2 segment would leave a file
2439	/// whose header promises a vocabulary its records exceed.
2440	#[test]
2441	fn an_old_segment_refuses_a_newer_operation() -> Outcome<()> {
2442	assert_eq!(highest_code(VERSION_MIN), crate::op::CODE_NOTE);
2443	assert_eq!(highest_code(3), crate::op::CODE_FILE_MODE);
2444	assert_eq!(highest_code(4), crate::op::CODE_REVERTS);
2445	assert_eq!(highest_code(5), crate::op::CODE_AMENDED);
2446	assert_eq!(highest_code(VERSION), crate::op::CODE_FORGOTTEN);
2447	// Every rung is named above, so a bump that forgot one would be caught here
2448	// rather than by a segment somebody could not read.
2449	assert!(highest_code(4) < highest_code(VERSION),
2450	"the vocabulary grows upwards, and version 5 must admit more than version 4");
2451	let old = Head { version: VERSION_MIN, replica: None };
2452	let mode = Entry::Bare(Record::root(
2453	oid(1, 1),
2454	Op::FileMode { file: oid(1, 1), mode: Mode::Symlink },
2455	));
2456	let mark = Entry::Bare(Record::root(oid(1, 2), Op::Mark { name: fmt!("v1"), body: None, time: None }));
2457	// Starting one.
2458	let mut writer: Writer<Fold, 0> = Writer::new(&old, Fold, [0u8; 0]);
2459	assert_eq!(writer.version(), VERSION_MIN);
2460	// An operation version 2 does spell goes in without complaint.
2461	res!(writer.push(&mark));
2462	let e = match writer.push(&mode) {
2463	Ok(()) => return Err(err!("A FileMode was written into a version 2 \
2464	segment."; Test)),
2465	Err(e) => e,
2466	};
2467	let msg = fmt!("{}", e);
2468	assert!(msg.contains("FileMode"), "message was {}", msg);
2469	assert!(msg.contains(&fmt!("{}", VERSION_MIN)), "message was {}", msg);
2470	// And continuing one, which is where a real repository would meet it.
2471	let bytes = res!(encode(&old, &[mark], Fold, [0u8; 0]));
2472	let mut writer: Writer<Fold, 0> = res!(Writer::resume(&bytes, Fold, [0u8; 0]));
2473	assert_eq!(writer.version(), VERSION_MIN);
2474	assert!(writer.push(&mode).is_err());
2475	// A segment at the current version takes it.
2476	let mut writer: Writer<Fold, 0> = Writer::new(&Head::new(None), Fold, [0u8; 0]);
2477	assert_eq!(writer.version(), VERSION);
2478	res!(writer.push(&mode));
2479	Ok(())
2480	}
2481
2482	/// A version 4 segment refuses an Amended, and a version 5 one takes it.
2483	///
2484	/// The same boundary as the version 3 test below, at the rung this change
2485	/// added, and it is worth its own test for the reason that one is: version 4
2486	/// is the version every repository written before this change is sitting in.
2487	/// Every existing store is therefore a version 4 store, and what a version 4
2488	/// segment does when handed code 14 is what decides whether an amendment costs
2489	/// anybody a migration. It does not -- the operation is refused by name and
2490	/// the caller opens a segment at the current version beside it.
2491	#[test]
2492	fn a_version_four_segment_refuses_an_amendment() -> Outcome<()> {
2493	let v4 = Head { version: 4, replica: None };
2494	let amended = Op::Amended {
2495	on: oid(3, 4),
2496	title: fmt!("Say it again"),
2497	body: b"and say it better".to_vec(),
2498	voice: fmt!("wren"),
2499	time: 1_755_400_300,
2500	};
2501	assert_eq!(amended.code(), crate::op::CODE_AMENDED);
2502	assert!(amended.code() > highest_code(4),
2503	"an amendment must sit above the version 4 vocabulary");
2504	let entry = Entry::Bare(Record::root(oid(9, 1), amended.clone()));
2505	// An operation version 4 does spell still goes into a version 4 segment, so
2506	// what follows is a refusal of this operation and not of the segment.
2507	let settled = Entry::Bare(Record::root(oid(9, 2), Op::Settled {
2508	on: oid(3, 4),
2509	state: crate::op::Settled::Accepted,
2510	mark: None,
2511	time: 1_755_400_301,
2512	}));
2513	let mut writer: Writer<Fold, 0> = Writer::new(&v4, Fold, [0u8; 0]);
2514	res!(writer.push(&settled));
2515	let e = match writer.push(&entry) {
2516	Ok(()) => return Err(err!(
2517	"An Amended at code {} was written into a version 4 segment.",
2518	amended.code(); Test)),
2519	Err(e) => e,
2520	};
2521	let msg = fmt!("{}", e);
2522	assert!(msg.contains("Amended"), "message was {}", msg);
2523	assert!(msg.contains(&fmt!("{}", amended.code())), "message was {}", msg);
2524	// And continuing a version 4 segment somebody else wrote, which is where a
2525	// real repository meets this rather than at a fresh one.
2526	let bytes = res!(encode(&v4, &[settled], Fold, [0u8; 0]));
2527	let mut writer: Writer<Fold, 0> = res!(Writer::resume(&bytes, Fold, [0u8; 0]));
2528	assert_eq!(writer.version(), 4);
2529	assert!(writer.push(&entry).is_err());
2530	// A segment at the current version takes it, and reads back what went in.
2531	let mut writer: Writer<Fold, 0> = Writer::new(&Head::new(None), Fold, [0u8; 0]);
2532	assert_eq!(writer.version(), VERSION);
2533	res!(writer.push(&entry));
2534	let back = res!(Op::from_dat(&amended.to_dat()));
2535	assert_eq!(back, amended, "an amendment did not survive its own encoding");
2536	Ok(())
2537	}
2538
2539	/// A version 3 segment refuses every operation version 4 added, and takes
2540	/// every operation version 3 spelled.
2541	///
2542	/// This is the mechanism the whole additive design rests on, at the boundary
2543	/// it was built for. Version 3 is the version every repository written before
2544	/// this change is sitting in, so what a version 3 segment does when it is
2545	/// handed a code above 8 is what decides whether the change costs a store a
2546	/// migration. It does not: the operation is refused, and the caller starts a
2547	/// segment at the current version rather than writing bytes into a file whose
2548	/// header promises a smaller vocabulary.
2549	#[test]
2550	fn a_version_three_segment_refuses_the_version_four_vocabulary() -> Outcome<()> {
2551	let v3 = Head { version: 3, replica: None };
2552	let newer = [
2553	// A mark carrying a time is the second spelling, at code 9.
2554	Op::Mark {
2555	name: fmt!("v1"),
2556	body: None,
2557	time: Some(1_755_000_000),
2558	},
2559	Op::Proposal {
2560	title: fmt!("Carry a body on a mark"),
2561	body: b"the case".to_vec(),
2562	voice: fmt!("someone"),
2563	time: 1_755_000_001,
2564	},
2565	Op::Said {
2566	on: oid(1, 1),
2567	text: b"agreed".to_vec(),
2568	voice: fmt!("someone else"),
2569	time: 1_755_000_002,
2570	},
2571	Op::Settled {
2572	on: oid(1, 1),
2573	state: crate::op::Settled::Accepted,
2574	mark: None,
2575	time: 1_755_000_003,
2576	},
2577	Op::Reverts { undone: vec![oid(1, 1), oid(2, 1)] },
2578	Op::Forget {
2579	of: vec![crate::op::Stub { id: oid(1, 1), placing: crate::op::Placing::File }],
2580	reason: b"a key".to_vec(),
2581	time: 1_755_000_004,
2582	},
2583	Op::Forgotten { placing: crate::op::Placing::Void },
2584	];
2585	for (i, op) in newer.iter().enumerate() {
2586	let code = op.code();
2587	assert!(code > highest_code(3), "{} is at code {}", op.name(), code);
2588	let entry = Entry::Bare(Record::root(oid(1, i as u64 + 1), op.clone()));
2589	let mut writer: Writer<Fold, 0> = Writer::new(&v3, Fold, [0u8; 0]);
2590	let e = match writer.push(&entry) {
2591	Ok(()) => return Err(err!(
2592	"A {} at code {} was written into a version 3 segment.",
2593	op.name(), code; Test)),
2594	Err(e) => e,
2595	};
2596	let msg = fmt!("{}", e);
2597	assert!(msg.contains(op.name()), "message was {}", msg);
2598	assert!(msg.contains(&fmt!("{}", code)), "message was {}", msg);
2599	// And a segment at the current version takes it.
2600	let mut writer: Writer<Fold, 0> = Writer::new(&Head::new(None), Fold, [0u8; 0]);
2601	res!(writer.push(&entry));
2602	}
2603	// The code the whole design turns on, said plainly: an operation at 13 in
2604	// a segment declaring 3.
2605	let reverts = Entry::Bare(Record::root(
2606	oid(9, 1),
2607	Op::Reverts { undone: vec![oid(1, 1)] },
2608	));
2609	assert_eq!(res!(reverts.peek()).op.code(), 13);
2610	let mut writer: Writer<Fold, 0> = Writer::new(&v3, Fold, [0u8; 0]);
2611	assert!(writer.push(&reverts).is_err());
2612	// A mark carrying neither a body nor a time is version 2 vocabulary and
2613	// goes into a version 3 segment, and a version 2 one, exactly as before.
2614	let plain = Entry::Bare(Record::root(
2615	oid(9, 2),
2616	Op::Mark { name: fmt!("v1"), body: None, time: None },
2617	));
2618	assert_eq!(res!(plain.peek()).op.code(), crate::op::CODE_MARK);
2619	let mut writer: Writer<Fold, 0> = Writer::new(&v3, Fold, [0u8; 0]);
2620	res!(writer.push(&plain));
2621	let old = Head { version: VERSION_MIN, replica: None };
2622	let mut writer: Writer<Fold, 0> = Writer::new(&old, Fold, [0u8; 0]);
2623	res!(writer.push(&plain));
2624	Ok(())
2625	}
2626
2627	/// The bytes of a one-record segment carrying a FileMode, frozen.
2628	///
2629	/// The operation that the version 3 bump exists for, pinned in the encoding
2630	/// it was added in on 12026-07-30. It is shaped like a FileRename -- a code,
2631	/// an identifier, a field -- and the field is a single tagged byte.
2632	///
2633	/// The version 4 and version 5 bumps each moved the version byte here and
2634	/// nothing else, which is the point: the operation this test pins was written
2635	/// in version 3 and is spelled in version 5 by the same bytes, so a segment
2636	/// full of them needs no migration.
2637	#[test]
2638	fn the_file_mode_bytes_are_frozen() -> Outcome<()> {
2639	let rec = Record::root(oid(1, 1), Op::FileMode {
2640	file: oid(1, 1),
2641	mode: Mode::Executable,
2642	});
2643	let bytes = res!(encode(&Head::new(None), &[Entry::Bare(rec)], Fold, [0u8; 0]));
2644	let want: &[u8] = &[
2645	// The magic, version 5, and no replica hint.
2646	0x4f, 0x52, 0x45, 0x53, 0x45, 0x47,
2647	0x06,
2648	0x00,
2649	// The record: a bare one, and 57 bytes of body.
2650	0x01,
2651	0x39,
2652	// The body is a daticle list of 54 bytes: the header, the operation.
2653	0x33, 0x21, 0x36,
2654	// The header, 23 bytes: the identifier r1:1, and no parents.
2655	0x33, 0x21, 0x17,
2656	0x33, 0x21, 0x12,
2657	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
2658	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
2659	0x33, 0x20,
2660	// The operation, 25 bytes: the FileMode code 8, the file it
2661	// names, and the mode, which is 1 for executable.
2662	0x33, 0x21, 0x19,
2663	0x0a, 0x08,
2664	0x33, 0x21, 0x12,
2665	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
2666	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
2667	0x0a, 0x01,
2668	// The digest: eight bytes of the folding hasher, and its length.
2669	0x08,
2670	0xd4, 0x4c, 0xdb, 0x41, 0x34, 0x92, 0x1e, 0xe8,
2671	];
2672	assert_eq!(bytes, want, "the FileMode encoding has changed");
2673	let (_, got) = res!(decode(want, Fold, [0u8; 0]));
2674	assert_eq!(got.len(), 1);
2675	match res!(got[0].peek()).op {
2676	Op::FileMode { file, mode } => {
2677	assert_eq!(file, oid(1, 1));
2678	assert_eq!(mode, Mode::Executable);
2679	},
2680	other => return Err(err!(
2681	"Expected a FileMode, got a {}.", other.name(); Test, Mismatch)),
2682	}
2683	Ok(())
2684	}
2685
2686	/// A replica hint of any size survives, and its absence is distinguishable
2687	/// from a hint of zero.
2688	#[test]
2689	fn the_replica_hint_round_trips() -> Outcome<()> {
2690	for hint in [None, Some(0u64), Some(1), Some(127), Some(128), Some(u64::MAX)] {
2691	let head = Head::new(hint.map(ReplicaId::new));
2692	let buf = head.encode();
2693	match res!(Head::decode(&buf)) {
2694	Some((got, used)) => {
2695	assert_eq!(got, head);
2696	assert_eq!(used, buf.len());
2697	},
2698	None => return Err(err!(
2699	"A whole header of {} bytes was read as a prefix.", buf.len();
2700	Test, Missing)),
2701	}
2702	}
2703	assert!(Head::new(None) != Head::new(Some(ReplicaId::new(0))));
2704	Ok(())
2705	}
2706
2707	#[test]
2708	fn the_writer_and_the_convenience_agree() -> Outcome<()> {
2709	let entries = res!(bare());
2710	let head = Head::new(Some(ReplicaId::new(4)));
2711	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
2712	for entry in &entries {
2713	res!(writer.push(entry));
2714	}
2715	assert_eq!(writer.count(), entries.len());
2716	assert_eq!(writer.finish(), res!(encode(&head, &entries, Fold, [0u8; 0])));
2717	Ok(())
2718	}
2719
2720	#[test]
2721	fn a_segment_resumes_where_it_left_off() -> Outcome<()> {
2722	let entries = res!(bare());
2723	let head = Head::new(Some(ReplicaId::new(11)));
2724	// The first go: a header and the first two records.
2725	let mut writer: Writer<Fold, 0> = Writer::new(&head, Fold, [0u8; 0]);
2726	res!(writer.extend(&entries[..2]));
2727	let mut file = writer.finish();
2728	// The second go, which starts by reading what is already there.
2729	let mut more: Writer<Fold, 0> = res!(Writer::resume(&file, Fold, [0u8; 0]));
2730	assert_eq!(more.count(), 2, "the records it was resumed from");
2731	res!(more.extend(&entries[2..]));
2732	assert_eq!(more.count(), entries.len());
2733	let tail = more.finish();
2734	file.extend_from_slice(&tail);
2735	// Which is the segment written in one go, byte for byte.
2736	assert_eq!(file, res!(encode(&head, &entries, Fold, [0u8; 0])));
2737	let (got_head, got) = res!(decode(&file, Fold, [0u8; 0]));
2738	assert_eq!(got_head, head);
2739	assert_eq!(got, entries);
2740	Ok(())
2741	}
2742
2743	#[test]
2744	fn an_empty_segment_resumes() -> Outcome<()> {
2745	let head = Head::new(None);
2746	let mut file = head.encode();
2747	let mut writer: Writer<Fold, 0> = res!(Writer::resume(&file, Fold, [0u8; 0]));
2748	assert_eq!(writer.count(), 0);
2749	let entries = res!(bare());
2750	res!(writer.push(&entries[0]));
2751	file.extend_from_slice(&writer.finish());
2752	let (_, got) = res!(decode(&file, Fold, [0u8; 0]));
2753	assert_eq!(got, entries[..1]);
2754	Ok(())
2755	}
2756
2757	/// Resuming under a hasher or a salt the segment was not written with is
2758	/// refused, because appending would leave a segment nobody could read whole.
2759	#[test]
2760	fn resuming_a_segment_written_otherwise_is_refused() -> Outcome<()> {
2761	let entries = res!(bare());
2762	let bytes = res!(encode(&Head::new(None), &entries, Fold, [0u8; 0]));
2763	assert!(Writer::<Fold, 4>::resume(&bytes, Fold, [1u8; 4]).is_err(), "a different salt");
2764	assert!(Writer::<(), 0>::resume(&bytes, (), [0u8; 0]).is_err(), "a different function");
2765	// A segment left half-written by an interrupted append, which is the
2766	// failure a resumed writer exists to avoid compounding. Every cut that
2767	// is not a record boundary is refused, and every cut that is one is a
2768	// shorter segment and resumes as such.
2769	let mut ends: Vec<usize> = Vec::new();
2770	let mut probe: Writer<Fold, 0> = Writer::new(&Head::new(None), Fold, [0u8; 0]);
2771	ends.push(probe.bytes().len());
2772	for entry in &entries {
2773	res!(probe.push(entry));
2774	ends.push(probe.bytes().len());
2775	}
2776	for cut in 1..bytes.len() {
2777	match ends.iter().position(\|e\| *e == cut) {
2778	Some(n) => {
2779	let w: Writer<Fold, 0> = res!(Writer::resume(&bytes[..cut], Fold, [0u8; 0]));
2780	assert_eq!(w.count(), n, "a segment of {} records cut at {}", n, cut);
2781	},
2782	None => if Writer::<Fold, 0>::resume(&bytes[..cut], Fold, [0u8; 0]).is_ok() {
2783	return Err(err!(
2784	"A segment cut at {} of {}, part way through a record, was \
2785	resumed.", cut, bytes.len();
2786	Test, Mismatch));
2787	},
2788	}
2789	}
2790	// And what is not a segment at all, including nothing.
2791	assert!(Writer::<Fold, 0>::resume(b"", Fold, [0u8; 0]).is_err());
2792	assert!(Writer::<Fold, 0>::resume(b"not a segment", Fold, [0u8; 0]).is_err());
2793	let mut wrong = MAGIC.to_vec();
2794	wrong.push(VERSION + 1);
2795	wrong.push(0);
2796	assert!(Writer::<Fold, 0>::resume(&wrong, Fold, [0u8; 0]).is_err(), "another version");
2797	Ok(())
2798	}
2799
2800	#[test]
2801	fn random_segments_round_trip() -> Outcome<()> {
2802	// A small linear congruential generator, so a failure can be reproduced.
2803	let mut state = 0x1234_5678_9abc_def0u64;
2804	let mut next = move \|\| {
2805	state = state
2806	.wrapping_mul(6_364_136_223_846_793_005)
2807	.wrapping_add(1_442_695_040_888_963_407);
2808	(state >> 33) as usize
2809	};
2810	let signer = StubSigner::with_seed(5);
2811	for trial in 0..40 {
2812	let n = next() % 12;
2813	let mut entries: Vec<Entry> = Vec::new();
2814	let mut ids: Vec<OpId> = Vec::new();
2815	for k in 0..n {
2816	let id = oid((next() % 5) as u64 + 1, k as u64 + 1);
2817	if ids.contains(&id) {
2818	continue;
2819	}
2820	let mut parents: Vec<OpId> = Vec::new();
2821	for cand in &ids {
2822	if next() % 2 == 0 {
2823	parents.push(*cand);
2824	}
2825	}
2826	let head = res!(Header::new(id, parents));
2827	let anchored = Some(Anchor::origin(oid((next() % 5) as u64 + 1, 1)));
2828	let op = match next() % 8 {
2829	0 => Op::FileCreate { path: fmt!("f{}", next() % 100).into_bytes() },
2830	1 => Op::Mark { name: fmt!("m{}", next() % 100), body: None, time: None },
2831	2 => Op::FileRename {
2832	file: oid((next() % 5) as u64 + 1, 1),
2833	path: vec![(next() % 256) as u8; next() % 40],
2834	},
2835	3 => Op::FileDelete { file: oid((next() % 5) as u64 + 1, 1) },
2836	4 => Op::Splice {
2837	left: anchored,
2838	right: None,
2839	remove: Vec::new(),
2840	insert: vec![(next() % 256) as u8; 1 + next() % 700].into(),
2841	},
2842	5 => Op::Move {
2843	src: vec![res!(ContentRange::new(id, 0, (next() % 50) as u64))],
2844	left: anchored,
2845	right: None,
2846	},
2847	6 => Op::FileMode {
2848	file: oid((next() % 5) as u64 + 1, 1),
2849	mode: match next() % 3 {
2850	0 => Mode::Normal,
2851	1 => Mode::Executable,
2852	_ => Mode::Symlink,
2853	},
2854	},
2855	_ => Op::Note {
2856	on: vec![res!(ContentRange::new(id, 0, (next() % 50) as u64 + 1))],
2857	text: fmt!("note {}", next() % 1000).into_bytes(),
2858	},
2859	};
2860	let rec = Record::new(head, op);
2861	entries.push(if next() % 3 == 0 {
2862	Entry::Sealed(res!(Envelope::seal_record(&signer, &rec)))
2863	} else {
2864	Entry::Bare(rec)
2865	});
2866	ids.push(id);
2867	}
2868	let head = Head::new(if next() % 2 == 0 {
2869	Some(ReplicaId::new(next() as u64))
2870	} else {
2871	None
2872	});
2873	let bytes = res!(encode(&head, &entries, Fold, [0u8; 0]));
2874	let (got_head, got) = res!(decode(&bytes, Fold, [0u8; 0]));
2875	assert_eq!(got_head, head, "trial {}", trial);
2876	assert_eq!(got, entries, "trial {}", trial);
2877	// And in arbitrary chunks.
2878	let mut reader: Reader<Fold, 0> = Reader::new(Fold, [0u8; 0]);
2879	let mut chunked: Vec<Entry> = Vec::new();
2880	let mut at = 0usize;
2881	while at < bytes.len() {
2882	let take = (1 + next() % 37).min(bytes.len() - at);
2883	reader.feed(&bytes[at..at + take]);
2884	at += take;
2885	while let Some(entry) = res!(reader.next_entry()) {
2886	chunked.push(entry);
2887	}
2888	}
2889	reader.end();
2890	while let Some(entry) = res!(reader.next_entry()) {
2891	chunked.push(entry);
2892	}
2893	assert_eq!(chunked, entries, "trial {} in chunks", trial);
2894	}
2895	Ok(())
2896	}
2897
2898	/// The bytes of a one-record segment, frozen.
2899	///
2900	/// A format that changes by accident orphans every store already written in
2901	/// it, and nothing else in this file would notice: every other test encodes
2902	/// and decodes with the same code. This one is the fixed point. If it fails
2903	/// and the change was deliberate, the version byte is the thing to raise.
2904	///
2905	/// It was raised to 2 for file identity, to 3 on 12026-07-30 for
2906	/// [`crate::op::Op::FileMode`], and to 4 on 12026-08-17 for the mark's second
2907	/// spelling and the proposal operations. The record below carries a mark with
2908	/// neither a body nor a time, whose encoding did not change on any of those
2909	/// occasions, so the only byte that has ever moved here is the version itself.
2910	/// That is the whole of each event as the framing sees it: what changed is
2911	/// which operations may appear inside, and an older segment stays readable
2912	/// because its operations are a subset of a newer one's.
2913	///
2914	/// The seven operation bytes below are therefore the test on the version 4
2915	/// design rather than a chore it creates. A mark saying nothing beyond its
2916	/// name is written at code 4 with two elements, as it always was; if it were
2917	/// re-spelled at code 9, those bytes would move and every mark ever signed
2918	/// would stop verifying.
2919	#[test]
2920	fn the_segment_bytes_are_frozen() -> Outcome<()> {
2921	let rec = Record::new(
2922	res!(Header::new(oid(2, 3), vec![oid(1, 7)])),
2923	Op::Mark { name: fmt!("v1"), body: None, time: None },
2924	);
2925	let bytes = res!(encode(
2926	&Head::new(Some(ReplicaId::new(2))),
2927	&[Entry::Bare(rec)],
2928	Fold,
2929	[0u8; 0],
2930	));
2931	let want: &[u8] = &[
2932	// The segment header: the magic, the version, a hint follows, and the
2933	// replica it names.
2934	0x4f, 0x52, 0x45, 0x53, 0x45, 0x47,
2935	0x06,
2936	0x01,
2937	0x02,
2938	// The record: a bare one, and 61 bytes of body.
2939	0x01,
2940	0x3d,
2941	// The body is a daticle list of 58 bytes: the header, the operation.
2942	0x33, 0x21, 0x3a,
2943	// The header, 45 bytes: the identifier, then the parents.
2944	0x33, 0x21, 0x2d,
2945	// The identifier r2:3, as two 64-bit integers.
2946	0x33, 0x21, 0x12,
2947	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02,
2948	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03,
2949	// One parent, r1:7.
2950	0x33, 0x21, 0x15,
2951	0x33, 0x21, 0x12,
2952	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
2953	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x07,
2954	// The operation, 7 bytes: the Mark code, and the name "v1".
2955	0x33, 0x21, 0x07,
2956	0x0a, 0x04,
2957	0x29, 0x21, 0x02, 0x76, 0x31,
2958	// The digest: eight bytes of the folding hasher, and its length.
2959	0x08,
2960	0x1e, 0x1a, 0xbf, 0xae, 0x11, 0xf1, 0xa0, 0xe5,
2961	];
2962	assert_eq!(bytes, want, "the segment format has changed");
2963	// Said again on its own: the mark's ten bytes, a three-byte list header
2964	// then the code 4 and the name, sitting where they have always sat. The
2965	// assertion above would catch them moving, but it would report a segment
2966	// that had changed rather than the thing that had actually gone wrong.
2967	let mark: &[u8] = &[0x33, 0x21, 0x07, 0x0a, 0x04, 0x29, 0x21, 0x02, 0x76, 0x31];
2968	assert!(
2969	bytes.windows(mark.len()).any(\|w\| w == mark),
2970	"a mark with neither a body nor a time is no longer written at code 4 \
2971	with two elements, so every mark ever signed has stopped verifying",
2972	);
2973	// And the frozen bytes still read.
2974	let (_, got) = res!(decode(want, Fold, [0u8; 0]));
2975	assert_eq!(got.len(), 1);
2976	assert_eq!(res!(got[0].id()), oid(2, 3));
2977	Ok(())
2978	}
2979
2980	/// The stand-in signer's keys are longer than a single byte length, so the
2981	/// sealed form exercises the wide byte fields.
2982	#[test]
2983	fn a_sealed_entry_carries_its_key() -> Outcome<()> {
2984	let (entries, _) = res!(sealed());
2985	match &entries[0] {
2986	Entry::Sealed(e) => assert!(e.signer().len() > 8),
2987	other => return Err(err!(
2988	"Expected a sealed envelope, got a {}.", other.name(); Test, Mismatch)),
2989	}
2990	let _ = StubSigner::default().clone_with_keys(None, None);
2991	Ok(())
2992	}
2993
2994	/// A veiled entry hands a carrier the header and nothing else, and gives a
2995	/// reader with the key back exactly what went in.
2996	///
2997	/// This is the whole of the form in one test. The identifier and the parents
2998	/// are readable without a key, because a carrier has to place the operation;
2999	/// the content is not, and is not in the segment's bytes to be found by
3000	/// searching for it; and what unveils is the same sealed envelope, whose
3001	/// signature still checks out, because nothing was re-encoded on the way.
3002	#[test]
3003	fn a_veiled_entry_carries_its_header_and_hides_the_rest() -> Outcome<()> {
3004	let signer = StubSigner::with_seed(3);
3005	let cipher = StubCipher::with_seed(11);
3006	let secret: &[u8] = b"the merger closes on Friday";
3007	let rec = Record::new(
3008	res!(Header::new(oid(2, 5), vec![oid(1, 3), oid(1, 4)])),
3009	Op::Splice {
3010	left: Some(Anchor::origin(oid(1, 3))),
3011	right: None,
3012	remove: Vec::new(),
3013	insert: secret.to_vec().into(),
3014	},
3015	);
3016	let plain = Entry::Sealed(res!(Envelope::seal_record(&signer, &rec)));
3017	let veiled = res!(plain.veil(&cipher));
3018	assert!(veiled.is_veiled());
3019	assert_eq!(veiled.kind(), KIND_VEILED);
3020	assert_eq!(veiled.name(), "veiled record");
3021
3022	// What a carrier may ask, and what it may not.
3023	let head = res!(veiled.head());
3024	assert_eq!(head.id(), oid(2, 5));
3025	assert_eq!(head.parents(), vec![oid(1, 3), oid(1, 4)]);
3026	assert_eq!(res!(veiled.id()), oid(2, 5));
3027	let refused = match veiled.peek() {
3028	Ok(_) => return Err(err!("A veiled operation was read."; Test, Security)),
3029	Err(e) => fmt!("{}", e.plain()),
3030	};
3031	assert!(refused.contains("r2:5"), "the refusal names the operation: {}", refused);
3032	assert!(refused.contains("veiled"), "and says what it met: {}", refused);
3033
3034	// Through a segment, which is what a carrier keeps.
3035	let bytes = res!(encode(&Head::new(None), &[veiled.clone()], Fold, [0u8; 0]));
3036	assert!(
3037	!bytes.windows(secret.len()).any(\|w\| w == secret),
3038	"the segment a carrier holds contains the operation's own content",
3039	);
3040	let (_, got) = res!(decode(&bytes, Fold, [0u8; 0]));
3041	assert_eq!(got, vec![veiled]);
3042
3043	// And a reader with the key gets back what was veiled, signature and all.
3044	let back = res!(got[0].unveil(&cipher));
3045	assert_eq!(back, plain);
3046	match &back {
3047	Entry::Sealed(env) => assert!(res!(env.verify(&signer)),
3048	"the signature made before veiling does not check out after"),
3049	other => return Err(err!(
3050	"Expected a sealed envelope, got a {}.", other.name(); Test, Mismatch)),
3051	}
3052	assert_eq!(res!(back.peek()), rec);
3053	Ok(())
3054	}
3055
3056	/// A carrier that rewrites the clear header is caught by the first reader
3057	/// holding the key, and told which copy to believe.
3058	///
3059	/// Rewriting it is the one thing a carrier can do to a veiled entry, and it is
3060	/// not nothing: every peer that never holds the key places the operation by the
3061	/// clear copy. What stops it mattering is that the copy inside is under the
3062	/// signature and cannot be made to agree.
3063	#[test]
3064	fn a_carrier_that_rewrites_the_clear_header_is_caught() -> Outcome<()> {
3065	let cipher = StubCipher::with_seed(2);
3066	let rec = Record::new(
3067	res!(Header::new(oid(1, 2), vec![oid(1, 1)])),
3068	Op::Mark { name: fmt!("v1"), body: None, time: None },
3069	);
3070	let veiled = res!(Entry::Bare(rec.clone()).veil(&cipher));
3071	assert_eq!(res!(veiled.unveil(&cipher)), Entry::Bare(rec), "sound as it stands");
3072	// Re-parented in clear, the ciphertext untouched.
3073	let lying = match &veiled {
3074	Entry::Veiled(v) => Entry::Veiled(Veiled {
3075	head: Header::root(oid(1, 2)),
3076	body: v.body.clone(),
3077	}),
3078	other => return Err(err!(
3079	"Expected a veiled record, got a {}.", other.name(); Test, Mismatch)),
3080	};
3081	assert!(res!(lying.head()).parents().is_empty(),
3082	"which is the graph a carrier would have placed it in");
3083	let caught = match lying.unveil(&cipher) {
3084	Ok(_) => return Err(err!(
3085	"A rewritten clear header was accepted."; Test, Security)),
3086	Err(e) => fmt!("{}", e.plain()),
3087	};
3088	assert!(caught.contains("r1:1"),
3089	"the refusal names the parent that was dropped: {}", caught);
3090	assert!(caught.contains("believe"),
3091	"and says which of the two copies to believe: {}", caught);
3092	Ok(())
3093	}
3094
3095	/// A key that is not the one it was veiled under fails by name rather than
3096	/// returning rubbish.
3097	#[test]
3098	fn the_wrong_key_does_not_unveil() -> Outcome<()> {
3099	let ours = StubCipher::with_seed(1);
3100	let theirs = StubCipher::with_seed(2);
3101	let veiled = res!(Entry::Bare(Record::root(
3102	oid(1, 1),
3103	Op::FileCreate { path: b"notes.md".to_vec() },
3104	)).veil(&ours));
3105	let refused = match veiled.unveil(&theirs) {
3106	Ok(_) => return Err(err!("Another key unveiled it."; Test, Security)),
3107	Err(e) => fmt!("{}", e.plain()),
3108	};
3109	assert!(refused.contains("r1:1"), "the refusal names the operation: {}", refused);
3110	assert!(refused.contains("did not decrypt"), "and says what failed: {}", refused);
3111	Ok(())
3112	}
3113
3114	#[test]
3115	fn veiling_does_not_nest_and_unveiling_wants_a_veil() -> Outcome<()> {
3116	let cipher = StubCipher::with_seed(7);
3117	let plain = Entry::Bare(Record::root(
3118	oid(1, 1),
3119	Op::Mark { name: fmt!("v1"), body: None, time: None },
3120	));
3121	let veiled = res!(plain.veil(&cipher));
3122	assert!(veiled.veil(&cipher).is_err(), "a veil over a veil hides the header");
3123	assert!(plain.unveil(&cipher).is_err(), "a bare record has nothing to unveil");
3124	Ok(())
3125	}
3126
3127	/// Veiled entries sit beside plain ones in one segment, and each comes back as
3128	/// the form it went in as.
3129	///
3130	/// A repository does not become veiled all at once: what a replica veils is
3131	/// what it hands to a carrier, and the segments either end of that hop hold
3132	/// whatever they were given. So the three forms have to mix, in a segment and
3133	/// in the daticle form a sync message carries.
3134	#[test]
3135	fn a_segment_mixes_veiled_entries_with_plain_ones() -> Outcome<()> {
3136	let cipher = StubCipher::with_seed(23);
3137	let (mut entries, _) = res!(sealed());
3138	entries.extend(res!(bare()));
3139	let mut mixed: Vec<Entry> = Vec::new();
3140	for (i, entry) in entries.iter().enumerate() {
3141	mixed.push(if i % 2 == 0 {
3142	res!(entry.veil(&cipher))
3143	} else {
3144	entry.clone()
3145	});
3146	}
3147	let bytes = res!(encode(&Head::new(None), &mixed, Fold, [0u8; 0]));
3148	let (_, got) = res!(decode(&bytes, Fold, [0u8; 0]));
3149	assert_eq!(got, mixed);
3150	for (i, entry) in got.iter().enumerate() {
3151	assert_eq!(&res!(Entry::from_dat(&entry.to_dat())), entry,
3152	"entry {} did not round trip as a daticle", i);
3153	assert_eq!(res!(entry.id()), res!(entries[i].id()),
3154	"entry {} is not the operation it was made from", i);
3155	let back = if entry.is_veiled() {
3156	res!(entry.unveil(&cipher))
3157	} else {
3158	entry.clone()
3159	};
3160	assert_eq!(back, entries[i], "entry {} did not come back as itself", i);
3161	}
3162	Ok(())
3163	}
3164
3165	/// The bytes of a one-record segment carrying a veiled entry, frozen.
3166	///
3167	/// Written under the identity encrypter, so what the array pins is the framing
3168	/// and not somebody's cipher: the kind byte 3, the header in clear, and a
3169	/// length prefixed body whose bytes here are the inner entry itself and can be
3170	/// read in the listing. Under a real cipher everything above the body is
3171	/// identical and the body is noise of the same length plus whatever the scheme
3172	/// adds to it.
3173	///
3174	/// The inner bytes are the same ten that [`the_segment_bytes_are_frozen`] pins
3175	/// for a mark, at their own offset, which is the point of veiling the tagged
3176	/// form rather than a re-encoding of it: what a reader with the key gets back
3177	/// is the entry that was signed, byte for byte.
3178	#[test]
3179	fn the_veiled_bytes_are_frozen() -> Outcome<()> {
3180	let rec = Record::new(
3181	res!(Header::new(oid(2, 3), vec![oid(1, 7)])),
3182	Op::Mark { name: fmt!("v1"), body: None, time: None },
3183	);
3184	let veiled = res!(Entry::Bare(rec).veil(&()));
3185	let bytes = res!(encode(&Head::new(None), &[veiled], Fold, [0u8; 0]));
3186	let want: &[u8] = &[
3187	// The magic, version 5, and no replica hint.
3188	0x4f, 0x52, 0x45, 0x53, 0x45, 0x47,
3189	0x06,
3190	0x00,
3191	// The record: a veiled one, and 126 bytes of body.
3192	0x03,
3193	0x7e,
3194	// The body is a daticle list of 123 bytes: the clear header, the
3195	// ciphertext.
3196	0x33, 0x21, 0x7b,
3197	// The header in clear, 45 bytes: the identifier r2:3 and the one
3198	// parent r1:7. This is the whole of what a carrier is given.
3199	0x33, 0x21, 0x2d,
3200	0x33, 0x21, 0x12,
3201	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02,
3202	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03,
3203	0x33, 0x21, 0x15,
3204	0x33, 0x21, 0x12,
3205	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
3206	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x07,
3207	// The body, as bytes under a 64-bit length: 66 of them. The length
3208	// is that wide because an operation is not bounded by 255 bytes and
3209	// a narrower field would decide the format by the first small one.
3210	0x47,
3211	0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x42,
3212	// Which under the identity encrypter is the inner entry itself,
3213	// tagged bare and carrying the record whole: the same header
3214	// again, then the mark at code 4 with the name "v1".
3215	0x33, 0x21, 0x3f,
3216	0x0a, 0x01,
3217	0x33, 0x21, 0x3a,
3218	0x33, 0x21, 0x2d,
3219	0x33, 0x21, 0x12,
3220	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02,
3221	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03,
3222	0x33, 0x21, 0x15,
3223	0x33, 0x21, 0x12,
3224	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
3225	0x0d, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x07,
3226	0x33, 0x21, 0x07,
3227	0x0a, 0x04,
3228	0x29, 0x21, 0x02, 0x76, 0x31,
3229	// The digest: eight bytes of the folding hasher, and its length.
3230	0x08,
3231	0xf5, 0x26, 0x35, 0x5c, 0x1f, 0x94, 0xed, 0x5a,
3232	];
3233	assert_eq!(bytes, want, "the veiled framing has changed");
3234	// The mark's own ten bytes, sitting inside the ciphertext at their usual
3235	// offset, which is what veiling the tagged form rather than a re-encoding
3236	// of it buys: a signature made before the veil holds after it.
3237	let mark: &[u8] = &[0x33, 0x21, 0x07, 0x0a, 0x04, 0x29, 0x21, 0x02, 0x76, 0x31];
3238	assert!(
3239	bytes.windows(mark.len()).any(\|w\| w == mark),
3240	"the entry a veil is put around is no longer the entry that was written",
3241	);
3242	// And the frozen bytes still read, and still unveil.
3243	let (_, got) = res!(decode(want, Fold, [0u8; 0]));
3244	assert_eq!(got.len(), 1);
3245	assert!(got[0].is_veiled());
3246	assert_eq!(res!(got[0].id()), oid(2, 3));
3247	assert_eq!(res!(res!(got[0].unveil(&())).peek()).head.parents(), vec![oid(1, 7)]);
3248	Ok(())
3249	}
3250
3251	/// Every shape an entry takes, measured and then encoded, and the two numbers
3252	/// compared.
3253	///
3254	/// This is the only test [`Entry::dat_len`] can have. It is a claim about
3255	/// bytes nobody built, and a carrier that believes it one byte short puts a
3256	/// reply past a bound it published -- which is a proxy closing a connection
3257	/// rather than a number being slightly wrong. So the corpus is every
3258	/// operation variant, in each of the three entry forms, and the payload sizes
3259	/// that move the compact length prefixes: nothing at all, one byte, and the
3260	/// 22,153,680 byte operation fe2o3's own history holds.
3261	///
3262	/// Proved red by adding one to the answer, and again by measuring the body
3263	/// alone rather than the tagged form a message carries, which is the very
3264	/// confusion between the two forms the doc above warns about.
3265	#[test]
3266	fn dat_len_is_what_the_entry_encodes_to() -> Outcome<()> {
3267	let signer = StubSigner::with_seed(3);
3268	let cipher = StubCipher::with_seed(9);
3269	let head = res!(Header::new(oid(5, 7), vec![oid(1, 1), oid(2, 2)]));
3270
3271	let mut ops: Vec<(String, Op)> = samples()
3272	.into_iter()
3273	.enumerate()
3274	.map(\|(i, op)\| (fmt!("sample {} ({})", i, op.name()), op))
3275	.collect();
3276	let payloads: [(&str, usize); 3] = [
3277	("an empty payload", 0),
3278	("a one byte payload", 1),
3279	("the 22,153,680 byte one", 22_153_680),
3280	];
3281	for (name, len) in payloads {
3282	ops.push((name.to_string(), Op::Splice {
3283	left: Some(Anchor::origin(oid(1, 1))),
3284	right: None,
3285	remove: Vec::new(),
3286	insert: vec![0x5a; len].into(),
3287	}));
3288	}
3289	assert!(ops.len() > 30, "the corpus is {} operations, which is not every shape", ops.len());
3290
3291	let mut seen = std::collections::BTreeSet::new();
3292	for (name, op) in ops {
3293	seen.insert(op.code());
3294	let rec = Record::new(head.clone(), op);
3295	let bare = Entry::Bare(rec.clone());
3296	let sealed = Entry::Sealed(res!(Envelope::seal_record(&signer, &rec)));
3297	let veiled = res!(bare.veil(&cipher));
3298	for (form, entry) in [("bare", bare), ("sealed", sealed), ("veiled", veiled)] {
3299	let said = res!(entry.dat_len());
3300	let wrote = res!(entry.to_dat().to_bytes(Vec::new())).len();
3301	assert_eq!(said, wrote,
3302	"{}, {}: measured at {} bytes and encoded to {}", name, form, said, wrote);
3303	}
3304	}
3305	assert_eq!(seen.len(), highest_code(VERSION) as usize,
3306	"the corpus covers {} of the {} operation codes this version writes",
3307	seen.len(), highest_code(VERSION));
3308	Ok(())
3309	}
3310	}