10. A content-addressed artifact store

Before a job can run on a machine, the machine needs two large things: the container image with the job’s programs, and usually a model’s weights. A Thor or a Spark has room for a model of a hundred gigabytes, and a training run produces a new version of such a model every few hours. Something has to hold all of these versions, deliver them to the machines, and do so without copying a hundred gigabytes every time one layer of a model changes.

In this chapter, we will build the storage scheme that container registries and model stores share. We will first see what is wrong with naming a file by a name someone chose, and what changes when a file is named by its own contents. Then we will write cas, in Go and in Rust, a store that keeps artifacts as chunks named by their hashes, and use it to push two versions of a model, pull one to a machine that already has the other, detect a corrupted chunk and repair it, and reclaim space.

What a chosen name does not promise

Suppose we store model weights in files named by hand: llm-v1.bin, llm-v2.bin, and llm-latest.bin. A machine that is told to load llm-latest.bin and has a file of that name faces three questions that the name cannot answer.

• Is it the right version? latest pointed at one model yesterday and points at another today. Two machines in the same training job can hold different files under the same name, and the job’s results depend on which machine a worker landed on.
• Is it intact? A file that was cut short by a full disk or changed by a failing memory chip still has its name.
• Do I already have it? If llm-v2.bin is llm-v1.bin with one layer retrained, the two files are almost the same bytes, and nothing in their names says so. The machine downloads all of it.

All three come from one cause: the name and the contents are unrelated, so knowing one tells us nothing about the other.

A name computed from the contents

A cryptographic hash function takes any sequence of bytes and computes a short, fixed-size value from it. SHA-256, the one we will use, produces 256 bits, written as 64 hexadecimal digits. Two properties make it useful here. The same bytes always give the same value. And although different inputs that give the same value must exist, nobody knows how to find two, so in practice a value identifies one particular sequence of bytes. The value is called a digest.

A content-addressed store uses the digest as the name. To store a blob of bytes, we compute its digest and file the blob under it. To fetch a blob, we ask for a digest. This one change answers all three questions.

• A digest names exactly one sequence of bytes, for ever. Two machines that load the same digest load the same model.
• Anyone who receives a blob can compute its digest and compare it with the name they asked for. A blob that was cut short or altered does not match.
• Two blobs with the same contents have the same name, so storing the second one is a lookup that finds the first. A machine that is asked for a digest it holds has nothing to download.

The same scheme on our node

The container images on our cluster are stored this way already. containerd keeps the parts of every image in a directory of files named by their SHA-256 digests. On our kind node:

$ docker exec gpu-lab-control-plane sh -c \
    'ls /var/lib/containerd/io.containerd.content.v1.content/blobs/sha256 | head -3'
0286df745178d6f77c036891bed8ddc93a624c1fa9c054d134674a096ee882e6
069d1e267530c2e681fbd4d481553b4d05f98082b18fafac86e7f12996dddd0b
0771a8a05619f4bb6f4c8484f942f509b21a04ebe56ee26d17a3006e250bfd46
$ docker exec gpu-lab-control-plane sh -c \
    'ls /var/lib/containerd/io.containerd.content.v1.content/blobs/sha256 | wc -l'
168

The node holds 168 blobs. The images we built in earlier chapters appear with the first digits of their digests when we list them:

$ docker exec gpu-lab-control-plane crictl images --digests | grep -E 'IMAGE|gpu-'
IMAGE                            TAG   DIGEST          IMAGE ID        SIZE
docker.io/library/gpu-check      1     3714eb7096522   07ee9035514b4   73.3MB
docker.io/library/gpu-sharing    1     413391f612a31   4212abdefc4b5   74.3MB

gpu-check and gpu-sharing were built from the same base image, so most of their 73 MB are the same blobs, stored once. A pod’s spec can name an image by its digest, gpu-check@sha256:..., and the kubelet then runs exactly those bytes on every node.

Chunks and manifests

Naming a whole 100 GB file by one digest answers the first two questions and does little for the third: change one byte and the digest changes, and the whole file is a new blob. To share the unchanged parts between versions, we store a large artifact in pieces.

• The artifact is cut into chunks of a fixed size. Each chunk is a blob, stored under its own digest.
• A manifest is a small text file that lists the chunks’ digests in order. It is a blob too, and its digest names the whole artifact: a different chunk anywhere gives a different line in the manifest and so a different manifest digest.
• A tag is a name a person chose, such as llm:v2, that points at a manifest’s digest. Tags are the only part of the store that can change.

Figure 10.1 shows two versions of a model stored this way.

Version 2 is version 1 with its last quarter retrained. Its first six chunks have the same bytes as version 1’s, therefore the same digests, therefore they are the same blobs. Pushing version 2 stores two new chunks and one new manifest.

Container images have had this shape for years. An image is a manifest that lists layers, each layer is a blob named by its digest, and two images built from the same base share the base’s layers. The format is standardised as the OCI image specification, and registries that store images this way are now also used to store model weights as OCI artifacts.

Building cas

Our store keeps its blobs in memory, in a map from digest to bytes, so that the program has no files to clean up. A store on disk is the same map with one file per blob, in a directory layout like the one containerd showed us.

Blobs by digest

Put computes the digest, and stores the bytes only if that digest is not already a key. Get does more than a lookup: it hashes the bytes again and compares the result with the name it was asked for. Every read is therefore a check of the data.

Listing 10.1 Digests, the store, and verified reads, in Go. code/go/cas/main.go

// Digest names a blob: "sha256:" followed by the hash of its bytes in hex.
type Digest string

// digestOf computes the name of some bytes.
func digestOf(data []byte) Digest {
	sum := sha256.Sum256(data)

	return Digest("sha256:" + hex.EncodeToString(sum[:]))
}

// short returns the first twelve hex digits, for printing.
func (d Digest) short() string { return string(d)[:19] }

// Store holds blobs by digest and tags by name.
type Store struct {
	// blobs maps each digest to the bytes it names.
	blobs map[Digest][]byte
	// tags maps a name people use, such as "llm:v1", to a manifest's digest.
	tags map[string]Digest
}

// NewStore returns an empty store.
func NewStore() *Store {
	return &Store{blobs: map[Digest][]byte{}, tags: map[string]Digest{}}
}

// Put stores a blob and returns its digest and whether it was new. Bytes the
// store already holds are not stored a second time.
func (s *Store) Put(data []byte) (Digest, bool) {
	d := digestOf(data)

	if _, held := s.blobs[d]; held {
		return d, false
	}

	s.blobs[d] = data

	return d, true
}

// ErrCorrupt is returned when a blob's bytes no longer match its name.
var ErrCorrupt = errors.New("contents do not match the digest")

// Get returns a blob after checking that its bytes still hash to its name.
func (s *Store) Get(d Digest) ([]byte, error) {
	data, held := s.blobs[d]

	if !held {
		return nil, fmt.Errorf("%s: not found", d.short())
	}

	if digestOf(data) != d {
		return nil, fmt.Errorf("%s: %w", d.short(), ErrCorrupt)
	}

	return data, nil
}

Listing 10.2 Digests, the store, and verified reads, in Rust. code/rust/cas/src/main.rs

/// Names a blob: "sha256:" followed by the hash of its bytes in hex.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
struct Digest(String);

impl Digest {
    /// Computes the name of some bytes.
    fn of(data: &[u8]) -> Self {
        let mut name = String::from("sha256:");

        for byte in Sha256::digest(data) {
            write!(name, "{byte:02x}").expect("writing to a String");
        }

        Digest(name)
    }

    /// Returns the first twelve hex digits, for printing.
    fn short(&self) -> &str {
        &self.0[..19]
    }
}

/// The ways reading from a store can fail.
#[derive(Debug)]
enum StoreError {
    /// The store holds no blob of this name.
    NotFound(String),
    /// A blob's bytes no longer match its name.
    Corrupt(String),
    /// A chunk of an artifact could not be read.
    Chunk(usize, Box<StoreError>),
}

impl std::fmt::Display for StoreError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            StoreError::NotFound(d) => write!(f, "{d}: not found"),
            StoreError::Corrupt(d) => write!(f, "{d}: contents do not match the digest"),
            StoreError::Chunk(i, e) => write!(f, "chunk {i}: {e}"),
        }
    }
}

/// Holds blobs by digest and tags by name.
#[derive(Default)]
struct Store {
    /// Maps each digest to the bytes it names.
    blobs: BTreeMap<Digest, Vec<u8>>,
    /// Maps a name people use, such as "llm:v1", to a manifest's digest.
    tags: BTreeMap<&'static str, Digest>,
}

impl Store {
    /// Stores a blob and returns its digest and whether it was new. Bytes the
    /// store already holds are not stored a second time.
    fn put(&mut self, data: &[u8]) -> (Digest, bool) {
        let d = Digest::of(data);

        if self.blobs.contains_key(&d) {
            return (d, false);
        }

        self.blobs.insert(d.clone(), data.to_vec());

        (d, true)
    }

    /// Returns a blob after checking that its bytes still hash to its name.
    fn get(&self, d: &Digest) -> Result<&[u8], StoreError> {
        let data = self
            .blobs
            .get(d)
            .ok_or_else(|| StoreError::NotFound(d.short().to_owned()))?;
        if Digest::of(data) != *d {
            return Err(StoreError::Corrupt(d.short().to_owned()));
        }

        Ok(data)
    }
}

The Go standard library includes SHA-256 in crypto/sha256. Rust’s standard library has no hash functions of this kind, and the Rust version uses the sha2 crate. The two programs print the same digests, which is itself a check: both computed the same function over the same bytes.

Pushing an artifact

Push walks the artifact in steps of the chunk size, stores each chunk with Put, and writes one line of the manifest for it. It counts how many chunks Put reported as new, which is the amount of data a push to a remote registry would have to send. The manifest is then stored like any other blob, and the tag is pointed at it.

Listing 10.3 Chunking an artifact and writing its manifest, in Go. code/go/cas/main.go

// chunkSize is the size of the pieces an artifact is cut into.
const chunkSize = 8 * 1024

// Push cuts an artifact into chunks, stores each chunk, stores a manifest
// that lists the chunks' digests in order, and points the tag at the
// manifest. It returns how many chunks were new to the store.
func (s *Store) Push(tag string, artifact []byte) (manifest Digest, chunks, added int) {
	var lines strings.Builder

	for start := 0; start < len(artifact); start += chunkSize {
		chunk := artifact[start:min(start+chunkSize, len(artifact))]
		d, fresh := s.Put(chunk)

		if fresh {
			added++
		}

		chunks++
		fmt.Fprintf(&lines, "%s %d\n", d, len(chunk))
	}

	manifest, _ = s.Put([]byte(lines.String()))
	s.tags[tag] = manifest

	return manifest, chunks, added
}

// chunksOf reads a manifest and returns the digests it lists.
func (s *Store) chunksOf(manifest Digest) ([]Digest, error) {
	data, err := s.Get(manifest)

	if err != nil {
		return nil, err
	}

	var out []Digest

	for line := range strings.Lines(string(data)) {
		digest, _, _ := strings.Cut(line, " ")
		out = append(out, Digest(digest))
	}

	return out, nil
}

Listing 10.4 Chunking an artifact and writing its manifest, in Rust. code/rust/cas/src/main.rs

impl Store {
    /// Cuts an artifact into chunks, stores each chunk, stores a manifest that
    /// lists the chunks' digests in order, and points the tag at the manifest.
    /// Returns the manifest's digest, the number of chunks, and how many were
    /// new to the store.
    fn push(&mut self, tag: &'static str, artifact: &[u8]) -> (Digest, usize, usize) {
        let mut lines = String::new();
        let (mut chunks, mut added) = (0, 0);

        for chunk in artifact.chunks(CHUNK_SIZE) {
            let (d, fresh) = self.put(chunk);
            added += usize::from(fresh);
            chunks += 1;
            writeln!(lines, "{} {}", d.0, chunk.len()).expect("writing to a String");
        }

        let (manifest, _) = self.put(lines.as_bytes());
        self.tags.insert(tag, manifest.clone());

        (manifest, chunks, added)
    }

    /// Reads a manifest and returns the digests it lists.
    fn chunks_of(&self, manifest: &Digest) -> Result<Vec<Digest>, StoreError> {
        let text = String::from_utf8_lossy(self.get(manifest)?);
        Ok(text
            .lines()
            .filter_map(|line| line.split(' ').next())
            .map(|d| Digest(d.to_owned()))
            .collect())
    }
}

Our chunks are 8 KiB so that the example stays small. Real stores use chunks of a few megabytes: large enough that the list of digests stays short, small enough that a change touches few of them.

Pulling and reading

Pull is how a machine gets an artifact. It reads the manifest from the remote store, and for each chunk it lists, fetches the chunk only if the local store has no blob of that name. Read reassembles the artifact from local chunks, and since it goes through Get, a damaged chunk stops the read with an error that says which chunk it was.

Listing 10.5 Fetching what is missing, and reading with verification, in Go. code/go/cas/main.go

// Pull copies a tagged artifact from a remote store into this one, fetching
// only the chunks this store does not hold. It returns the number of chunks
// and bytes fetched.
func (s *Store) Pull(remote *Store, tag string) (fetched, bytes int, err error) {
	manifest := remote.tags[tag]
	chunks, err := remote.chunksOf(manifest)

	if err != nil {
		return 0, 0, err
	}

	for _, d := range chunks {
		if _, held := s.blobs[d]; held {
			continue
		}

		data, err := remote.Get(d)

		if err != nil {
			return fetched, bytes, err
		}

		s.blobs[d] = data
		fetched++
		bytes += len(data)
	}

	data, _ := remote.Get(manifest)
	s.blobs[manifest] = data
	s.tags[tag] = manifest

	return fetched, bytes, nil
}

// Read reassembles a tagged artifact, verifying every chunk on the way.
func (s *Store) Read(tag string) ([]byte, error) {
	chunks, err := s.chunksOf(s.tags[tag])

	if err != nil {
		return nil, err
	}

	var artifact []byte

	for i, d := range chunks {
		data, err := s.Get(d)

		if err != nil {
			return nil, fmt.Errorf("chunk %d: %w", i, err)
		}

		artifact = append(artifact, data...)
	}

	return artifact, nil
}

Listing 10.6 Fetching what is missing, and reading with verification, in Rust. code/rust/cas/src/main.rs

impl Store {
    /// Copies a tagged artifact from a remote store into this one, fetching
    /// only the chunks this store does not hold. Returns the number of chunks
    /// and bytes fetched.
    fn pull(&mut self, remote: &Store, tag: &'static str) -> Result<(usize, usize), StoreError> {
        let manifest = &remote.tags[tag];
        let (mut fetched, mut bytes) = (0, 0);

        for d in remote.chunks_of(manifest)? {
            if self.blobs.contains_key(&d) {
                continue;
            }

            let data = remote.get(&d)?;
            fetched += 1;
            bytes += data.len();
            self.blobs.insert(d, data.to_vec());
        }

        self.blobs
            .insert(manifest.clone(), remote.get(manifest)?.to_vec());
        self.tags.insert(tag, manifest.clone());

        Ok((fetched, bytes))
    }

    /// Reassembles a tagged artifact, verifying every chunk on the way.
    fn read(&self, tag: &str) -> Result<Vec<u8>, StoreError> {
        let mut artifact = Vec::new();

        for (i, d) in self.chunks_of(&self.tags[tag])?.iter().enumerate() {
            let data = self.get(d).map_err(|e| StoreError::Chunk(i, Box::new(e)))?;
            artifact.extend_from_slice(data);
        }

        Ok(artifact)
    }
}

Reclaiming space

A content-addressed store grows with every push, and blobs cannot be deleted one artifact at a time, because a chunk may be listed by several manifests. The store removes blobs by garbage collection in two phases. The mark phase starts from the tags, which are the only names people use, and notes every manifest they point at and every chunk those manifests list. The sweep phase removes every blob that was not noted.

Listing 10.7 Mark from the tags, then sweep, in Go. code/go/cas/main.go

// Collect removes every blob that no tag leads to. It first marks the
// manifests the tags name and the chunks those manifests list, then sweeps
// the rest. It returns the number of blobs and bytes removed.
func (s *Store) Collect() (removed, bytes int) {
	live := map[Digest]bool{}

	for _, manifest := range s.tags {
		live[manifest] = true
		chunks, _ := s.chunksOf(manifest)

		for _, d := range chunks {
			live[d] = true
		}
	}

	for d, data := range s.blobs {
		if !live[d] {
			removed++
			bytes += len(data)
			delete(s.blobs, d)
		}
	}

	return removed, bytes
}

Listing 10.8 Mark from the tags, then sweep, in Rust. code/rust/cas/src/main.rs

impl Store {
    /// Removes every blob that no tag leads to. It first marks the manifests
    /// the tags name and the chunks those manifests list, then sweeps the
    /// rest. Returns the number of blobs and bytes removed.
    fn collect(&mut self) -> (usize, usize) {
        let mut live = BTreeSet::new();

        for manifest in self.tags.values() {
            live.extend(self.chunks_of(manifest).unwrap_or_default());
            live.insert(manifest.clone());
        }

        let (before, bytes_before) = self.size();
        self.blobs.retain(|d, _| live.contains(d));
        let (after, bytes_after) = self.size();

        (before - after, bytes_before - bytes_after)
    }
}

The run

The model’s weights are 64 KiB from a fixed pseudo-random sequence, which stand in for gigabytes. Version 2 keeps the first 48 KiB and replaces the last 16.

Listing 10.9 Two versions, a pull, a damaged chunk, and a collection, in Go. code/go/cas/main.go

func main() {
	// Version 1 is 64 KiB. Version 2 is a fine-tuned copy: the first three
	// quarters are unchanged and the last quarter is new.
	v1 := weights(64*1024, 1)
	v2 := append(append([]byte{}, v1[:48*1024]...), weights(16*1024, 2)...)

	registry := NewStore()
	m1, chunks, added := registry.Push("llm:v1", v1)
	fmt.Printf("push llm:v1: %d chunks, %d new; manifest %s\n", chunks, added, m1.short())
	m2, chunks, added := registry.Push("llm:v2", v2)
	fmt.Printf("push llm:v2: %d chunks, %d new; manifest %s\n", chunks, added, m2.short())
	blobs, bytes := registry.size()
	fmt.Printf("the registry holds %d blobs and %d bytes for %d bytes of artifacts\n", blobs, bytes, len(v1)+len(v2))

	node := NewStore()
	fetched, n, _ := node.Pull(registry, "llm:v1")
	fmt.Printf("thor-1 pulls llm:v1: %d chunks fetched, %d bytes\n", fetched, n)
	fetched, n, _ = node.Pull(registry, "llm:v2")
	fmt.Printf("thor-1 pulls llm:v2: %d chunks fetched, %d bytes\n", fetched, n)

	fmt.Println("one bit flips in a chunk stored on thor-1:")
	chunkList, _ := node.chunksOf(node.tags["llm:v1"])
	damaged := append([]byte{}, node.blobs[chunkList[3]]...)
	damaged[100] ^= 1
	node.blobs[chunkList[3]] = damaged

	if _, err := node.Read("llm:v1"); err != nil {
		fmt.Printf("  read llm:v1: %v\n", err)
	}

	// The bad chunk is dropped and pulled again, by name, from the registry.
	delete(node.blobs, chunkList[3])
	fetched, n, _ = node.Pull(registry, "llm:v1")
	fmt.Printf("  thor-1 drops the chunk and pulls llm:v1 again: %d chunk fetched, %d bytes\n", fetched, n)

	if got, err := node.Read("llm:v1"); err == nil {
		fmt.Printf("  read llm:v1: %d bytes, digest %s\n", len(got), digestOf(got).short())
	}

	fmt.Println("the tag llm:v1 is deleted from the registry:")
	delete(registry.tags, "llm:v1")
	removed, n := registry.Collect()
	blobs, bytes = registry.size()
	fmt.Printf("  collected %d blobs, %d bytes; %d blobs and %d bytes remain\n", removed, n, blobs, bytes)
}

Listing 10.10 Two versions, a pull, a damaged chunk, and a collection, in Rust. code/rust/cas/src/main.rs

fn main() -> Result<(), StoreError> {
    // Version 1 is 64 KiB. Version 2 is a fine-tuned copy: the first three
    // quarters are unchanged and the last quarter is new.
    let v1 = weights(64 * 1024, 1);
    let v2 = [&v1[..48 * 1024], &weights(16 * 1024, 2)].concat();

    let mut registry = Store::default();
    let (m1, chunks, added) = registry.push("llm:v1", &v1);
    println!(
        "push llm:v1: {chunks} chunks, {added} new; manifest {}",
        m1.short()
    );
    let (m2, chunks, added) = registry.push("llm:v2", &v2);
    println!(
        "push llm:v2: {chunks} chunks, {added} new; manifest {}",
        m2.short()
    );
    let (blobs, bytes) = registry.size();
    println!(
        "the registry holds {blobs} blobs and {bytes} bytes for {} bytes of artifacts",
        v1.len() + v2.len()
    );

    let mut node = Store::default();
    let (fetched, n) = node.pull(&registry, "llm:v1")?;
    println!("thor-1 pulls llm:v1: {fetched} chunks fetched, {n} bytes");
    let (fetched, n) = node.pull(&registry, "llm:v2")?;
    println!("thor-1 pulls llm:v2: {fetched} chunks fetched, {n} bytes");

    println!("one bit flips in a chunk stored on thor-1:");
    let fourth = node.chunks_of(&node.tags["llm:v1"])?[3].clone();
    node.blobs.get_mut(&fourth).expect("the chunk was pulled")[100] ^= 1;

    if let Err(e) = node.read("llm:v1") {
        println!("  read llm:v1: {e}");
    }

    // The bad chunk is dropped and pulled again, by name, from the registry.
    node.blobs.remove(&fourth);
    let (fetched, n) = node.pull(&registry, "llm:v1")?;
    println!("  thor-1 drops the chunk and pulls llm:v1 again: {fetched} chunk fetched, {n} bytes");
    let got = node.read("llm:v1")?;
    println!(
        "  read llm:v1: {} bytes, digest {}",
        got.len(),
        Digest::of(&got).short()
    );

    println!("the tag llm:v1 is deleted from the registry:");
    registry.tags.remove("llm:v1");
    let (removed, n) = registry.collect();
    let (blobs, bytes) = registry.size();
    println!("  collected {removed} blobs, {n} bytes; {blobs} blobs and {bytes} bytes remain");

    Ok(())
}

$ go run ./cas
push llm:v1: 8 chunks, 8 new; manifest sha256:eea2b9042521
push llm:v2: 8 chunks, 2 new; manifest sha256:7df8dd2bef7e
the registry holds 12 blobs and 83152 bytes for 131072 bytes of artifacts
thor-1 pulls llm:v1: 8 chunks fetched, 65536 bytes
thor-1 pulls llm:v2: 2 chunks fetched, 16384 bytes
one bit flips in a chunk stored on thor-1:
  read llm:v1: chunk 3: sha256:156fd5154cde: contents do not match the digest
  thor-1 drops the chunk and pulls llm:v1 again: 1 chunk fetched, 8192 bytes
  read llm:v1: 65536 bytes, digest sha256:a6a98d7f309f
the tag llm:v1 is deleted from the registry:
  collected 3 blobs, 17000 bytes; 9 blobs and 66152 bytes remain

Pushing llm:v1 stores eight chunks. Pushing llm:v2 cuts the new artifact into eight chunks as well and finds six of them already in the store, so it adds two. The registry holds 12 blobs, ten chunks and two manifests, in 83,152 bytes, for two artifacts that total 131,072 bytes.

thor-1 starts empty and fetches all eight chunks of version 1. For version 2 it fetches two chunks, 16,384 bytes, a quarter of the artifact. On a 100 GB model with the same proportion changed, that is 25 GB where a store of whole files would send 100, and at the 3.125 GB per second of one 25 GbE lane it is 8 seconds where the whole file takes 32.

Then one bit of one chunk on thor-1 flips. The next read of llm:v1 stops at chunk 3 and names the digest whose contents no longer match. The repair is as local as the damage: the machine drops that one blob and pulls again, the pull finds seven chunks present and one missing, and 8,192 bytes come from the registry. The digest printed for the reassembled artifact, sha256:a6a98d7f309f, is the digest of the whole 64 KiB of version 1.

Finally the tag llm:v1 is deleted. Collection marks the manifest of version 2 and its eight chunks, and sweeps three blobs: the manifest of version 1 and the two chunks that only version 1 listed. The six shared chunks stay, because version 2 still leads to them.

Animation 10.1 plays the run, with the registry’s chunks on the left and those on thor-1 on the right.

What the control plane gains

Content addressing is useful to every part of the system built in the earlier chapters.

A GpuJob that names its image and its model by digest runs the same bytes on every worker of a gang, which chapter 7’s synchronous step depends on. The checkpoint store of chapter 4 can be one of these: a checkpoint is an artifact, its chunks that did not change since the last checkpoint are not written again, and the epoch check sits on the tag, which is the only thing that is overwritten. And a machine that returns from a reimage, the last rung of chapter 4’s ladder, refills its local store by pulling, and the verification on every chunk doubles as a test of its new disk.

The complete program

Listing 10.11 The complete program in Go. code/go/cas/main.go

// Command cas builds a small content-addressed store for model artifacts and
// runs it through the situations such a store exists for.
//
// A blob is stored under the SHA-256 digest of its bytes. An artifact is cut
// into chunks, each chunk is a blob, and a manifest lists the chunks. The
// program pushes two versions of a model that differ in their last quarter,
// pulls the second to a machine that has the first, corrupts one stored
// chunk, and removes what nothing refers to any more.
//
// Run it from code/go with:
//
//	go run ./cas
package main

import (
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
)

// Digest names a blob: "sha256:" followed by the hash of its bytes in hex.
type Digest string

// digestOf computes the name of some bytes.
func digestOf(data []byte) Digest {
	sum := sha256.Sum256(data)

	return Digest("sha256:" + hex.EncodeToString(sum[:]))
}

// short returns the first twelve hex digits, for printing.
func (d Digest) short() string { return string(d)[:19] }

// Store holds blobs by digest and tags by name.
type Store struct {
	// blobs maps each digest to the bytes it names.
	blobs map[Digest][]byte
	// tags maps a name people use, such as "llm:v1", to a manifest's digest.
	tags map[string]Digest
}

// NewStore returns an empty store.
func NewStore() *Store {
	return &Store{blobs: map[Digest][]byte{}, tags: map[string]Digest{}}
}

// Put stores a blob and returns its digest and whether it was new. Bytes the
// store already holds are not stored a second time.
func (s *Store) Put(data []byte) (Digest, bool) {
	d := digestOf(data)

	if _, held := s.blobs[d]; held {
		return d, false
	}

	s.blobs[d] = data

	return d, true
}

// ErrCorrupt is returned when a blob's bytes no longer match its name.
var ErrCorrupt = errors.New("contents do not match the digest")

// Get returns a blob after checking that its bytes still hash to its name.
func (s *Store) Get(d Digest) ([]byte, error) {
	data, held := s.blobs[d]

	if !held {
		return nil, fmt.Errorf("%s: not found", d.short())
	}

	if digestOf(data) != d {
		return nil, fmt.Errorf("%s: %w", d.short(), ErrCorrupt)
	}

	return data, nil
}


// chunkSize is the size of the pieces an artifact is cut into.
const chunkSize = 8 * 1024

// Push cuts an artifact into chunks, stores each chunk, stores a manifest
// that lists the chunks' digests in order, and points the tag at the
// manifest. It returns how many chunks were new to the store.
func (s *Store) Push(tag string, artifact []byte) (manifest Digest, chunks, added int) {
	var lines strings.Builder

	for start := 0; start < len(artifact); start += chunkSize {
		chunk := artifact[start:min(start+chunkSize, len(artifact))]
		d, fresh := s.Put(chunk)

		if fresh {
			added++
		}

		chunks++
		fmt.Fprintf(&lines, "%s %d\n", d, len(chunk))
	}

	manifest, _ = s.Put([]byte(lines.String()))
	s.tags[tag] = manifest

	return manifest, chunks, added
}

// chunksOf reads a manifest and returns the digests it lists.
func (s *Store) chunksOf(manifest Digest) ([]Digest, error) {
	data, err := s.Get(manifest)

	if err != nil {
		return nil, err
	}

	var out []Digest

	for line := range strings.Lines(string(data)) {
		digest, _, _ := strings.Cut(line, " ")
		out = append(out, Digest(digest))
	}

	return out, nil
}


// Pull copies a tagged artifact from a remote store into this one, fetching
// only the chunks this store does not hold. It returns the number of chunks
// and bytes fetched.
func (s *Store) Pull(remote *Store, tag string) (fetched, bytes int, err error) {
	manifest := remote.tags[tag]
	chunks, err := remote.chunksOf(manifest)

	if err != nil {
		return 0, 0, err
	}

	for _, d := range chunks {
		if _, held := s.blobs[d]; held {
			continue
		}

		data, err := remote.Get(d)

		if err != nil {
			return fetched, bytes, err
		}

		s.blobs[d] = data
		fetched++
		bytes += len(data)
	}

	data, _ := remote.Get(manifest)
	s.blobs[manifest] = data
	s.tags[tag] = manifest

	return fetched, bytes, nil
}

// Read reassembles a tagged artifact, verifying every chunk on the way.
func (s *Store) Read(tag string) ([]byte, error) {
	chunks, err := s.chunksOf(s.tags[tag])

	if err != nil {
		return nil, err
	}

	var artifact []byte

	for i, d := range chunks {
		data, err := s.Get(d)

		if err != nil {
			return nil, fmt.Errorf("chunk %d: %w", i, err)
		}

		artifact = append(artifact, data...)
	}

	return artifact, nil
}


// Collect removes every blob that no tag leads to. It first marks the
// manifests the tags name and the chunks those manifests list, then sweeps
// the rest. It returns the number of blobs and bytes removed.
func (s *Store) Collect() (removed, bytes int) {
	live := map[Digest]bool{}

	for _, manifest := range s.tags {
		live[manifest] = true
		chunks, _ := s.chunksOf(manifest)

		for _, d := range chunks {
			live[d] = true
		}
	}

	for d, data := range s.blobs {
		if !live[d] {
			removed++
			bytes += len(data)
			delete(s.blobs, d)
		}
	}

	return removed, bytes
}


// size returns the number of blobs in a store and their total size.
func (s *Store) size() (blobs, bytes int) {
	for _, data := range s.blobs {
		bytes += len(data)
	}

	return len(s.blobs), bytes
}

// weights returns n bytes that stand for a model's weights. The bytes come
// from a fixed sequence, so every run and both languages produce the same.
func weights(n int, seed uint32) []byte {
	out := make([]byte, n)
	x := seed

	for i := range out {
		x = x*1664525 + 1013904223
		out[i] = byte(x >> 24)
	}

	return out
}

func main() {
	// Version 1 is 64 KiB. Version 2 is a fine-tuned copy: the first three
	// quarters are unchanged and the last quarter is new.
	v1 := weights(64*1024, 1)
	v2 := append(append([]byte{}, v1[:48*1024]...), weights(16*1024, 2)...)

	registry := NewStore()
	m1, chunks, added := registry.Push("llm:v1", v1)
	fmt.Printf("push llm:v1: %d chunks, %d new; manifest %s\n", chunks, added, m1.short())
	m2, chunks, added := registry.Push("llm:v2", v2)
	fmt.Printf("push llm:v2: %d chunks, %d new; manifest %s\n", chunks, added, m2.short())
	blobs, bytes := registry.size()
	fmt.Printf("the registry holds %d blobs and %d bytes for %d bytes of artifacts\n", blobs, bytes, len(v1)+len(v2))

	node := NewStore()
	fetched, n, _ := node.Pull(registry, "llm:v1")
	fmt.Printf("thor-1 pulls llm:v1: %d chunks fetched, %d bytes\n", fetched, n)
	fetched, n, _ = node.Pull(registry, "llm:v2")
	fmt.Printf("thor-1 pulls llm:v2: %d chunks fetched, %d bytes\n", fetched, n)

	fmt.Println("one bit flips in a chunk stored on thor-1:")
	chunkList, _ := node.chunksOf(node.tags["llm:v1"])
	damaged := append([]byte{}, node.blobs[chunkList[3]]...)
	damaged[100] ^= 1
	node.blobs[chunkList[3]] = damaged

	if _, err := node.Read("llm:v1"); err != nil {
		fmt.Printf("  read llm:v1: %v\n", err)
	}

	// The bad chunk is dropped and pulled again, by name, from the registry.
	delete(node.blobs, chunkList[3])
	fetched, n, _ = node.Pull(registry, "llm:v1")
	fmt.Printf("  thor-1 drops the chunk and pulls llm:v1 again: %d chunk fetched, %d bytes\n", fetched, n)

	if got, err := node.Read("llm:v1"); err == nil {
		fmt.Printf("  read llm:v1: %d bytes, digest %s\n", len(got), digestOf(got).short())
	}

	fmt.Println("the tag llm:v1 is deleted from the registry:")
	delete(registry.tags, "llm:v1")
	removed, n := registry.Collect()
	blobs, bytes = registry.size()
	fmt.Printf("  collected %d blobs, %d bytes; %d blobs and %d bytes remain\n", removed, n, blobs, bytes)
}

Listing 10.12 The complete program in Rust. code/rust/cas/src/main.rs

//! Builds a small content-addressed store for model artifacts and runs it
//! through the situations such a store exists for.
//!
//! A blob is stored under the SHA-256 digest of its bytes. An artifact is cut
//! into chunks, each chunk is a blob, and a manifest lists the chunks. The
//! program pushes two versions of a model that differ in their last quarter,
//! pulls the second to a machine that has the first, corrupts one stored
//! chunk, and removes what nothing refers to any more.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p cas
//! ```

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Write as _;

use sha2::{Digest as _, Sha256};

/// Names a blob: "sha256:" followed by the hash of its bytes in hex.
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
struct Digest(String);

impl Digest {
    /// Computes the name of some bytes.
    fn of(data: &[u8]) -> Self {
        let mut name = String::from("sha256:");

        for byte in Sha256::digest(data) {
            write!(name, "{byte:02x}").expect("writing to a String");
        }

        Digest(name)
    }

    /// Returns the first twelve hex digits, for printing.
    fn short(&self) -> &str {
        &self.0[..19]
    }
}

/// The ways reading from a store can fail.
#[derive(Debug)]
enum StoreError {
    /// The store holds no blob of this name.
    NotFound(String),
    /// A blob's bytes no longer match its name.
    Corrupt(String),
    /// A chunk of an artifact could not be read.
    Chunk(usize, Box<StoreError>),
}

impl std::fmt::Display for StoreError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            StoreError::NotFound(d) => write!(f, "{d}: not found"),
            StoreError::Corrupt(d) => write!(f, "{d}: contents do not match the digest"),
            StoreError::Chunk(i, e) => write!(f, "chunk {i}: {e}"),
        }
    }
}

/// Holds blobs by digest and tags by name.
#[derive(Default)]
struct Store {
    /// Maps each digest to the bytes it names.
    blobs: BTreeMap<Digest, Vec<u8>>,
    /// Maps a name people use, such as "llm:v1", to a manifest's digest.
    tags: BTreeMap<&'static str, Digest>,
}

impl Store {
    /// Stores a blob and returns its digest and whether it was new. Bytes the
    /// store already holds are not stored a second time.
    fn put(&mut self, data: &[u8]) -> (Digest, bool) {
        let d = Digest::of(data);

        if self.blobs.contains_key(&d) {
            return (d, false);
        }

        self.blobs.insert(d.clone(), data.to_vec());

        (d, true)
    }

    /// Returns a blob after checking that its bytes still hash to its name.
    fn get(&self, d: &Digest) -> Result<&[u8], StoreError> {
        let data = self
            .blobs
            .get(d)
            .ok_or_else(|| StoreError::NotFound(d.short().to_owned()))?;
        if Digest::of(data) != *d {
            return Err(StoreError::Corrupt(d.short().to_owned()));
        }

        Ok(data)
    }
}

impl Store {
    /// Cuts an artifact into chunks, stores each chunk, stores a manifest that
    /// lists the chunks' digests in order, and points the tag at the manifest.
    /// Returns the manifest's digest, the number of chunks, and how many were
    /// new to the store.
    fn push(&mut self, tag: &'static str, artifact: &[u8]) -> (Digest, usize, usize) {
        let mut lines = String::new();
        let (mut chunks, mut added) = (0, 0);

        for chunk in artifact.chunks(CHUNK_SIZE) {
            let (d, fresh) = self.put(chunk);
            added += usize::from(fresh);
            chunks += 1;
            writeln!(lines, "{} {}", d.0, chunk.len()).expect("writing to a String");
        }

        let (manifest, _) = self.put(lines.as_bytes());
        self.tags.insert(tag, manifest.clone());

        (manifest, chunks, added)
    }

    /// Reads a manifest and returns the digests it lists.
    fn chunks_of(&self, manifest: &Digest) -> Result<Vec<Digest>, StoreError> {
        let text = String::from_utf8_lossy(self.get(manifest)?);
        Ok(text
            .lines()
            .filter_map(|line| line.split(' ').next())
            .map(|d| Digest(d.to_owned()))
            .collect())
    }
}

impl Store {
    /// Copies a tagged artifact from a remote store into this one, fetching
    /// only the chunks this store does not hold. Returns the number of chunks
    /// and bytes fetched.
    fn pull(&mut self, remote: &Store, tag: &'static str) -> Result<(usize, usize), StoreError> {
        let manifest = &remote.tags[tag];
        let (mut fetched, mut bytes) = (0, 0);

        for d in remote.chunks_of(manifest)? {
            if self.blobs.contains_key(&d) {
                continue;
            }

            let data = remote.get(&d)?;
            fetched += 1;
            bytes += data.len();
            self.blobs.insert(d, data.to_vec());
        }

        self.blobs
            .insert(manifest.clone(), remote.get(manifest)?.to_vec());
        self.tags.insert(tag, manifest.clone());

        Ok((fetched, bytes))
    }

    /// Reassembles a tagged artifact, verifying every chunk on the way.
    fn read(&self, tag: &str) -> Result<Vec<u8>, StoreError> {
        let mut artifact = Vec::new();

        for (i, d) in self.chunks_of(&self.tags[tag])?.iter().enumerate() {
            let data = self.get(d).map_err(|e| StoreError::Chunk(i, Box::new(e)))?;
            artifact.extend_from_slice(data);
        }

        Ok(artifact)
    }
}

impl Store {
    /// Removes every blob that no tag leads to. It first marks the manifests
    /// the tags name and the chunks those manifests list, then sweeps the
    /// rest. Returns the number of blobs and bytes removed.
    fn collect(&mut self) -> (usize, usize) {
        let mut live = BTreeSet::new();

        for manifest in self.tags.values() {
            live.extend(self.chunks_of(manifest).unwrap_or_default());
            live.insert(manifest.clone());
        }

        let (before, bytes_before) = self.size();
        self.blobs.retain(|d, _| live.contains(d));
        let (after, bytes_after) = self.size();

        (before - after, bytes_before - bytes_after)
    }
}

impl Store {
    /// Returns the number of blobs in the store and their total size.
    fn size(&self) -> (usize, usize) {
        (self.blobs.len(), self.blobs.values().map(Vec::len).sum())
    }
}

/// The size of the pieces an artifact is cut into.
const CHUNK_SIZE: usize = 8 * 1024;

/// Returns n bytes that stand for a model's weights. The bytes come from a
/// fixed sequence, so every run and both languages produce the same.
fn weights(n: usize, seed: u32) -> Vec<u8> {
    let mut x = seed;
    (0..n)
        .map(|_| {
            x = x.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
            (x >> 24) as u8
        })
        .collect()
}

fn main() -> Result<(), StoreError> {
    // Version 1 is 64 KiB. Version 2 is a fine-tuned copy: the first three
    // quarters are unchanged and the last quarter is new.
    let v1 = weights(64 * 1024, 1);
    let v2 = [&v1[..48 * 1024], &weights(16 * 1024, 2)].concat();

    let mut registry = Store::default();
    let (m1, chunks, added) = registry.push("llm:v1", &v1);
    println!(
        "push llm:v1: {chunks} chunks, {added} new; manifest {}",
        m1.short()
    );
    let (m2, chunks, added) = registry.push("llm:v2", &v2);
    println!(
        "push llm:v2: {chunks} chunks, {added} new; manifest {}",
        m2.short()
    );
    let (blobs, bytes) = registry.size();
    println!(
        "the registry holds {blobs} blobs and {bytes} bytes for {} bytes of artifacts",
        v1.len() + v2.len()
    );

    let mut node = Store::default();
    let (fetched, n) = node.pull(&registry, "llm:v1")?;
    println!("thor-1 pulls llm:v1: {fetched} chunks fetched, {n} bytes");
    let (fetched, n) = node.pull(&registry, "llm:v2")?;
    println!("thor-1 pulls llm:v2: {fetched} chunks fetched, {n} bytes");

    println!("one bit flips in a chunk stored on thor-1:");
    let fourth = node.chunks_of(&node.tags["llm:v1"])?[3].clone();
    node.blobs.get_mut(&fourth).expect("the chunk was pulled")[100] ^= 1;

    if let Err(e) = node.read("llm:v1") {
        println!("  read llm:v1: {e}");
    }

    // The bad chunk is dropped and pulled again, by name, from the registry.
    node.blobs.remove(&fourth);
    let (fetched, n) = node.pull(&registry, "llm:v1")?;
    println!("  thor-1 drops the chunk and pulls llm:v1 again: {fetched} chunk fetched, {n} bytes");
    let got = node.read("llm:v1")?;
    println!(
        "  read llm:v1: {} bytes, digest {}",
        got.len(),
        Digest::of(&got).short()
    );

    println!("the tag llm:v1 is deleted from the registry:");
    registry.tags.remove("llm:v1");
    let (removed, n) = registry.collect();
    let (blobs, bytes) = registry.size();
    println!("  collected {removed} blobs, {n} bytes; {blobs} blobs and {bytes} bytes remain");

    Ok(())
}

References

• Open Container Initiative, Image Format Specification: https://github.com/opencontainers/image-spec/blob/main/spec.md
• Open Container Initiative, Distribution Specification: https://github.com/opencontainers/distribution-spec/blob/main/spec.md
• containerd, Content flow: https://github.com/containerd/containerd/blob/main/docs/content-flow.md
• Kubernetes, Images: https://kubernetes.io/docs/concepts/containers/images/
• NIST, FIPS 180-4, Secure Hash Standard: https://csrc.nist.gov/pubs/fips/180-4/upd1/final