Machine learning systems in production face a bootstrapping problem: before the application can serve its first request, model weights must be fetched from a remote registry, validated, placed in the correct directory, and the inference engine must confirm readiness. In search systems that use embedding models, there’s an additional dependency—vector indexes must be created or rebuilt whenever the underlying model changes, because different models produce incompatible embedding spaces.
A well-designed search engine implements a model registry puller that fetches manifests from a model registry (e.g., registry.example.io), streams model blobs into the local inference server’s models directory with real-time progress tracking, and orchestrates the lifecycle of embedding indexes. The engine waits for the local inference server to report healthy before creating or recreating text and image embedding indexes. When models or chunking strategies change, the system detects the mismatch and triggers a full reindex.
This architecture separates concerns cleanly: the registry puller handles artifact distribution (fetching, caching, integrity verification), the local inference server handles inference (loading models, serving embedding requests), and the search engine orchestrates the lifecycle (deciding when to pull, when to reindex, and how to handle failures). Each component can evolve independently as long as the contracts between them are maintained.
A model registry (e.g., registry.example.io) can use the OCI (Open Container Initiative) distribution spec to store model artifacts. Each model is a manifest containing references to one or more blobs (layers). The manifest includes metadata like model architecture, quantization format, input/output dimensions, and content-addressed digests for integrity verification. This is the same protocol that Docker registries use for container images, which means standard tooling (like crane, oras, or skopeo) can inspect and manage model artifacts.
Every model blob is identified by its SHA-256 digest, not by a mutable tag. When you pull registry.example.io/models/embedding-v3:latest, the tag resolves to a specific manifest digest, and each blob in the manifest has its own digest. This means pulls are idempotent (pulling the same digest twice is a no-op if the blob is cached), integrity is guaranteed (any corruption is detected by digest mismatch), and caching is trivial (store blobs by digest, never worry about stale data). Content-addressing eliminates an entire class of “wrong model version” bugs.
Model blobs can be hundreds of megabytes to several gigabytes. The puller streams blobs directly to disk rather than buffering in memory, reporting progress at configurable intervals. Progress callbacks enable the UI to show download percentage, transfer speed, and estimated time remaining. The streaming design also enables resume-on-failure: if a download is interrupted, the puller can resume from the last successfully written byte using HTTP range requests, avoiding re-downloading the entire blob.
The local inference server loads ONNX or safetensors models and exposes embedding and classification endpoints. After the puller places model files in the inference server’s models directory, the engine polls the inference server’s health endpoint until the model is loaded and ready to serve. This health gate prevents the search engine from attempting to create embedding indexes before the embedding model is available, which would produce empty or error-filled indexes.
Embedding vectors are meaningful only in the context of the model that produced them. Vectors from model v1 cannot be meaningfully compared with vectors from model v2, even if both produce the same dimensionality. When the model changes (new version pulled from registry), all existing embedding indexes must be invalidated and rebuilt. The search engine tracks the model digest alongside each index. If the current model’s digest doesn’t match the index’s recorded digest, the index is marked stale and scheduled for rebuilding.
Embedding indexes are also invalidated when the chunking strategy changes. A chunker splits documents into segments before embedding; changing the segment size, overlap, or splitting algorithm produces different text chunks, which produce different embeddings, which require a new index. The search engine stores a hash of the chunker configuration alongside the model digest in each index’s metadata. Either a model change or a chunker change triggers reindexing.
Index creation is a two-phase process. Phase 1 (build) creates the index in the background: documents are chunked, embedded via the inference server, and inserted into a new index instance. Phase 2 (swap) atomically replaces the old index with the new one. During Phase 1, the old index continues serving queries—there is no downtime. The swap is a pointer update: replace the active index reference, then schedule the old index for garbage collection after in-flight queries drain. This gives zero-downtime reindexing even for large document collections.
The system has a strict startup ordering: (1) pull model from registry, (2) wait for the inference server to load the model and become healthy, (3) check if existing indexes match the current model+chunker, (4) rebuild stale indexes, (5) begin serving queries. Each step depends on the previous step’s success. The engine models this as a dependency graph with timeouts at each stage. If the inference server doesn’t become healthy within the timeout, the engine logs a diagnostic error and enters a degraded mode that serves cached results without embedding-based search.
Fetch an OCI manifest from the model registry and resolve blob references:
package registry
import (
"context"
"encoding/json"
"fmt"
"net/http"
)
// Manifest describes a model artifact in the registry.
type Manifest struct {
SchemaVersion int `json:"schemaVersion"`
MediaType string `json:"mediaType"`
Config Blob `json:"config"`
Layers []Blob `json:"layers"`
}
// Blob is a content-addressed reference to a model artifact layer.
type Blob struct {
MediaType string `json:"mediaType"`
Digest string `json:"digest"` // e.g., "sha256:abc123..."
Size int64 `json:"size"`
}
// Puller fetches model artifacts from an OCI-compatible registry.
type Puller struct {
registryURL string
client *http.Client
modelsDir string
}
// FetchManifest retrieves the manifest for a model reference.
func (p *Puller) FetchManifest(ctx context.Context,
repo, tag string) (*Manifest, error) {
url := fmt.Sprintf("%s/v2/%s/manifests/%s", p.registryURL, repo, tag)
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return nil, err
}
req.Header.Set("Accept",
"application/vnd.oci.image.manifest.v1+json")
resp, err := p.client.Do(req)
if err != nil {
return nil, fmt.Errorf("fetching manifest: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("registry returned %d for %s:%s",
resp.StatusCode, repo, tag)
}
var manifest Manifest
if err := json.NewDecoder(resp.Body).Decode(&manifest); err != nil {
return nil, fmt.Errorf("decoding manifest: %w", err)
}
return &manifest, nil
}Stream model blobs to disk with real-time progress reporting:
// ProgressFunc reports download progress.
type ProgressFunc func(downloaded, total int64)
// PullBlob downloads a blob to the models directory with progress tracking.
func (p *Puller) PullBlob(ctx context.Context, repo string,
blob Blob, onProgress ProgressFunc) (string, error) {
// Check if blob already exists (content-addressed caching)
destPath := filepath.Join(p.modelsDir, digestToFilename(blob.Digest))
if info, err := os.Stat(destPath); err == nil {
if info.Size() == blob.Size {
return destPath, nil // Already cached
}
}
url := fmt.Sprintf("%s/v2/%s/blobs/%s",
p.registryURL, repo, blob.Digest)
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return "", err
}
// Support resume via Range header
tmpPath := destPath + ".partial"
var startByte int64
if info, err := os.Stat(tmpPath); err == nil {
startByte = info.Size()
req.Header.Set("Range",
fmt.Sprintf("bytes=%d-", startByte))
}
resp, err := p.client.Do(req)
if err != nil {
return "", fmt.Errorf("downloading blob: %w", err)
}
defer resp.Body.Close()
// Open file for append (resume) or create
flags := os.O_WRONLY | os.O_CREATE
if startByte > 0 && resp.StatusCode == http.StatusPartialContent {
flags |= os.O_APPEND
} else {
flags |= os.O_TRUNC
startByte = 0
}
f, err := os.OpenFile(tmpPath, flags, 0644)
if err != nil {
return "", err
}
defer f.Close()
// Stream with progress reporting
downloaded := startByte
buf := make([]byte, 256*1024) // 256KB buffer
for {
n, readErr := resp.Body.Read(buf)
if n > 0 {
if _, writeErr := f.Write(buf[:n]); writeErr != nil {
return "", fmt.Errorf("writing blob: %w", writeErr)
}
downloaded += int64(n)
if onProgress != nil {
onProgress(downloaded, blob.Size)
}
}
if readErr != nil {
if readErr == io.EOF {
break
}
return "", fmt.Errorf("reading blob: %w", readErr)
}
}
// Verify digest before finalizing
if err := verifyDigest(tmpPath, blob.Digest); err != nil {
os.Remove(tmpPath)
return "", fmt.Errorf("digest mismatch: %w", err)
}
// Atomic rename to final path
if err := os.Rename(tmpPath, destPath); err != nil {
return "", err
}
return destPath, nil
}Coordinate model readiness, index staleness detection, and zero-downtime reindexing:
// IndexManager orchestrates the embedding index lifecycle.
type IndexManager struct {
puller *Puller
inferenceServer *InferenceServer
indexStore *IndexStore
config IndexConfig
}
// IndexConfig specifies model and chunker settings for an index.
type IndexConfig struct {
ModelRepo string
ModelTag string
ChunkerHash string // Hash of chunker configuration
IndexType string // "text" or "image"
}
/// EnsureReady performs the full startup sequence:
// pull model → wait for inference server → check/rebuild indexes.
func (im *IndexManager) EnsureReady(ctx context.Context) error {
// Phase 1: Pull model from registry
manifest, err := im.puller.FetchManifest(ctx,
im.config.ModelRepo, im.config.ModelTag)
if err != nil {
return fmt.Errorf("fetching manifest: %w", err)
}
for _, layer := range manifest.Layers {
_, err := im.puller.PullBlob(ctx,
im.config.ModelRepo, layer,
func(dl, total int64) {
pct := float64(dl) / float64(total) * 100
log.Printf("Downloading %s: %.1f%%", layer.Digest[:12], pct)
},
)
if err != nil {
return fmt.Errorf("pulling blob %s: %w", layer.Digest[:12], err)
}
}
// Phase 2: Wait for the inference server to load the model
if err := im.waitForInferenceServer(ctx); err != nil {
return fmt.Errorf("inference server health check: %w", err)
}
// Phase 3: Check if indexes need rebuilding
modelDigest := manifest.Config.Digest
currentMeta := IndexMeta{
ModelDigest: modelDigest,
ChunkerHash: im.config.ChunkerHash,
}
existing, err := im.indexStore.GetMeta(im.config.IndexType)
if err != nil || existing.ModelDigest != currentMeta.ModelDigest ||
existing.ChunkerHash != currentMeta.ChunkerHash {
// Index is stale or missing — rebuild
log.Printf("Index stale: model %s→%s, chunker %s→%s",
existing.ModelDigest[:12], currentMeta.ModelDigest[:12],
existing.ChunkerHash[:8], currentMeta.ChunkerHash[:8])
return im.rebuildIndex(ctx, currentMeta)
}
log.Printf("Index up-to-date for %s", im.config.IndexType)
return nil
}
// waitForInferenceServer polls the health endpoint with exponential backoff.
func (im *IndexManager) waitForInferenceServer(ctx context.Context) error {
backoff := 500 * time.Millisecond
maxBackoff := 30 * time.Second
deadline := time.After(5 * time.Minute)
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-deadline:
return fmt.Errorf("inference server not ready after 5 minutes")
default:
}
if im.inferenceServer.IsHealthy(ctx) {
return nil
}
time.Sleep(backoff)
backoff = min(backoff*2, maxBackoff)
}
}During reindexing, the old index continues serving all queries. The new index is built in a separate directory. Once complete, the swap is an atomic pointer update: the active index reference changes from old to new. In-flight queries that started on the old index complete normally because the old index remains valid until its reference count drops to zero. Only then is the old index directory cleaned up. This pattern borrows from database MVCC: readers never block writers, and writers never block readers.
When a new embedding model version is published to the registry, the update process is: pull the new model blobs (which doesn’t affect the running model), signal the inference server to load the new model in parallel with the old one, rebuild indexes using the new model, swap indexes atomically, then unload the old model from the inference server. At no point during this sequence do search queries fail or return stale results.
The search engine manages separate indexes for text embeddings and image embeddings, each backed by a different model. The index manager runs the lifecycle independently for each model type. Text embedding indexes can be rebuilt while image embedding indexes continue serving, and vice versa. This independence means a slow image model pull doesn’t block text search availability.
For edge deployments with intermittent connectivity, the puller caches model blobs locally by digest. If the registry is unreachable at startup, the engine falls back to the most recently cached model version. The content-addressed storage means the cached version is guaranteed to be intact (verified by digest on last pull). Indexes built from the cached model continue to work; they’re only invalidated when a newer model actually downloads successfully.
When an operator changes the chunking strategy (e.g., increasing chunk size from 256 to 512 tokens), the system detects the chunker hash mismatch on the next startup or configuration reload. It automatically triggers a full reindex using the current model and new chunker settings, without requiring manual intervention. This makes configuration changes safe: the operator changes a value, and the system converges to the correct state.
Build a new index in the background and swap atomically:
// rebuildIndex creates a new index and swaps it with the old one.
func (im *IndexManager) rebuildIndex(ctx context.Context,
meta IndexMeta) error {
// Create a new index in a temporary directory
tmpDir, err := os.MkdirTemp(im.indexStore.BaseDir(), "reindex-*")
if err != nil {
return fmt.Errorf("creating temp dir: %w", err)
}
newIndex, err := NewEmbeddingIndex(tmpDir, im.config.IndexType)
if err != nil {
os.RemoveAll(tmpDir)
return fmt.Errorf("creating index: %w", err)
}
// Iterate all documents, chunk, embed, and insert
docs, err := im.indexStore.ListDocuments(ctx)
if err != nil {
return fmt.Errorf("listing documents: %w", err)
}
log.Printf("Rebuilding %s index: %d documents", im.config.IndexType, len(docs))
for i, doc := range docs {
chunks := im.chunker.Chunk(doc.Content)
embeddings, err := im.inferenceServer.EmbedBatch(ctx, chunks)
if err != nil {
return fmt.Errorf("embedding doc %s: %w", doc.ID, err)
}
for j, emb := range embeddings {
if err := newIndex.Insert(doc.ID, j, chunks[j], emb); err != nil {
return fmt.Errorf("inserting chunk: %w", err)
}
}
if (i+1)%100 == 0 {
log.Printf("Reindex progress: %d/%d documents", i+1, len(docs))
}
}
// Finalize the new index (build HNSW graph, etc.)
if err := newIndex.Finalize(); err != nil {
os.RemoveAll(tmpDir)
return fmt.Errorf("finalizing index: %w", err)
}
// Atomic swap: update active index pointer
oldDir := im.indexStore.Swap(im.config.IndexType, tmpDir, meta)
// Schedule old index cleanup after drain period
go func() {
time.Sleep(30 * time.Second) // Allow in-flight queries to complete
os.RemoveAll(oldDir)
log.Printf("Cleaned up old %s index", im.config.IndexType)
}()
log.Printf("Index rebuild complete for %s", im.config.IndexType)
return nil
}Verify downloaded blob integrity using content-addressed hashing:
// verifyDigest computes the SHA-256 digest of a file and compares
// it against the expected digest from the manifest.
func verifyDigest(path, expected string) error {
f, err := os.Open(path)
if err != nil {
return err
}
defer f.Close()
h := sha256.New()
if _, err := io.Copy(h, f); err != nil {
return fmt.Errorf("hashing file: %w", err)
}
actual := "sha256:" + hex.EncodeToString(h.Sum(nil))
if actual != expected {
return fmt.Errorf("expected %s, got %s", expected, actual)
}
return nil
}
// digestToFilename converts a digest to a safe filename.
// "sha256:abc123def456..." → "sha256-abc123def456..."
func digestToFilename(digest string) string {
return strings.Replace(digest, ":", "-", 1)
}Because blobs are stored by digest, pulling is inherently idempotent. If a blob with the requested digest already exists on disk and its size matches, the pull is a no-op. If a partial download exists (from a previous interrupted pull), the puller resumes from where it left off. If the partial file is corrupted, digest verification fails and the puller retries from scratch. This design means the puller can be called repeatedly without wasted work or corrupted state—exactly the property you want for a startup sequence that may be interrupted by crashes or restarts.
EmbedBatch call (32–64 chunks per batch) amortizes inference server round-trip latency. With a local inference server, expect 500–2000 embeddings/second depending on model size and hardware.