A stream processor becomes interesting the moment “read a record, write a record” is no longer enough. Inputs can be larger than RAM, processes can die between writes, malformed records can halt progress, and a graceful shutdown can arrive while work is in flight.

This tutorial builds a production-oriented Go processor for immutable JSON Lines files. It keeps application buffers bounded, writes batches durably, checkpoints only completed input, and recovers safely when a failure occurs between committing output and advancing the checkpoint.

The processor deliberately remains single-threaded. That preserves source order and makes the checkpoint the highest contiguous byte offset completed. Concurrency can improve CPU-heavy transformations, but it also requires a bounded reorder buffer and may leave later records waiting behind one slow record.

Prerequisites and guarantees

You need Go 1.22 or newer and a Linux filesystem whose atomic link, rename, and directory synchronization behavior you have validated. Local ext4 and XFS are typical deployment choices; network filesystems require explicit testing.

The input contract is strict:

• The source is a regular, immutable JSONL file.
• Every record ends with a newline.
• One record cannot exceed 1 MiB.
• The output directory belongs to one processing job.

The processor hashes the source before starting and stores its size and SHA-256 digest in the checkpoint. This costs one full read at startup, but prevents accidentally resuming a checkpoint against different input. For very large sources, immutable object identifiers or manifests can replace startup hashing.

Processing remains honestly at-least-once. Durable segment creation is idempotent, but any downstream database, API, or message publication must use its own idempotency key, such as the source digest plus source_offset.

Architecture and failure boundary

A buffered reader accepts one bounded record at a time. Transformed records accumulate until either 500 records or 4 MiB of encoded output is reached. Because reading stops while a batch is persisted, backpressure is automatic and memory does not grow with source size.

1. Read and validate a record.
2. Transform it deterministically.
3. Add it to the bounded batch.
4. Write the batch to a temporary file and call fsync.
5. Atomically link it to a filename derived from source offsets.
6. Synchronize the output directory.
7. Atomically replace and synchronize the checkpoint.

If the process dies after step five but before step seven, recovery regenerates the same filename and content. An existing identical segment is accepted. Different content at the same offsets is treated as corruption or a nondeterministic transformation.

Project structure

jsonstream/
├── cmd/
│   └── jsonstream/
│       └── main.go
├── deploy/
│   └── jsonstream.service
└── go.mod

Create go.mod without third-party dependencies:

module example.com/jsonstream

go 1.22

Implement the processor

The example input represents account events. The transformation validates identifiers and classifies each amount while retaining the original byte offset for downstream deduplication.

package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"
)

const (
	maxRecordBytes = 1 << 20
	maxBatchBytes  = 4 << 20
	maxBatchEvents = 500
)

type config struct {
	input, out, state string
	failAfterSegment bool
}

type event struct {
	ID          string `json:"id"`
	Account     string `json:"account"`
	AmountCents int64  `json:"amount_cents"`
}

type output struct {
	ID           string `json:"id"`
	Account      string `json:"account"`
	AmountCents  int64  `json:"amount_cents"`
	Class        string `json:"class"`
	SourceOffset int64  `json:"source_offset"`
}

type checkpoint struct {
	SourceSize   int64  `json:"source_size"`
	SourceSHA256 string `json:"source_sha256"`
	Offset       int64  `json:"offset"`
}

type batch struct {
	start, end int64
	count      int
	data       bytes.Buffer
}

func main() {
	var cfg config
	flag.StringVar(&cfg.input, "input", "events.jsonl", "immutable JSONL source")
	flag.StringVar(&cfg.out, "out", "run/out", "segment directory")
	flag.StringVar(&cfg.state, "state", "run/checkpoint.json", "checkpoint path")
	flag.BoolVar(&cfg.failAfterSegment, "fail-after-segment", false, "test recovery boundary")
	flag.Parse()

	logger := slog.New(slog.NewJSONHandler(os.Stderr, nil))
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	if err := run(ctx, cfg, logger); err != nil {
		logger.Error("stream_failed", "error", err)
		os.Exit(1)
	}
}

func run(ctx context.Context, cfg config, logger *slog.Logger) error {
	if err := os.MkdirAll(cfg.out, 0750); err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(cfg.state), 0750); err != nil {
		return err
	}

	f, err := os.Open(cfg.input)
	if err != nil {
		return err
	}
	defer f.Close()

	before, err := f.Stat()
	if err != nil || !before.Mode().IsRegular() {
		return fmt.Errorf("input must be a regular file")
	}

	h := sha256.New()
	if _, err := io.Copy(h, f); err != nil {
		return fmt.Errorf("hash input: %w", err)
	}
	after, err := f.Stat()
	if err != nil {
		return err
	}
	if before.Size() != after.Size() || !before.ModTime().Equal(after.ModTime()) {
		return fmt.Errorf("input changed while hashing")
	}
	digest := hex.EncodeToString(h.Sum(nil))

	cp, found, err := loadCheckpoint(cfg.state)
	if err != nil {
		return err
	}
	if !found {
		cp = checkpoint{SourceSize: after.Size(), SourceSHA256: digest}
		if err := saveCheckpoint(cfg.state, cp); err != nil {
			return err
		}
	} else if cp.SourceSize != after.Size() || cp.SourceSHA256 != digest {
		return fmt.Errorf("checkpoint belongs to different input")
	}
	if cp.Offset < 0 || cp.Offset > cp.SourceSize {
		return fmt.Errorf("invalid checkpoint offset %d", cp.Offset)
	}
	if _, err := f.Seek(cp.Offset, io.SeekStart); err != nil {
		return err
	}

	reader := bufio.NewReaderSize(f, maxRecordBytes+1)
	offset := cp.Offset
	var current *batch

	flush := func() error {
		if current == nil {
			return nil
		}
		if err := commitBatch(current, cfg, &cp, logger); err != nil {
			return err
		}
		current = nil
		return nil
	}

	for {
		if ctx.Err() != nil {
			return flush()
		}

		line, readErr := reader.ReadSlice('\n')

		// A stop arriving during the blocking read must not start new work.
		if ctx.Err() != nil {
			return flush()
		}
		if errors.Is(readErr, bufio.ErrBufferFull) {
			return fmt.Errorf("record at offset %d exceeds %d bytes", offset, maxRecordBytes)
		}
		if errors.Is(readErr, io.EOF) {
			if len(line) != 0 {
				return fmt.Errorf("final record at offset %d lacks newline", offset)
			}
			if err := flush(); err != nil {
				return err
			}
			logger.Info("stream_complete", "offset", cp.Offset, "source_size", cp.SourceSize)
			return nil
		}
		if readErr != nil {
			return fmt.Errorf("read at offset %d: %w", offset, readErr)
		}
		if len(line) > maxRecordBytes {
			return fmt.Errorf("record at offset %d exceeds %d bytes", offset, maxRecordBytes)
		}

		encoded, err := transform(line, offset)
		if err != nil {
			return fmt.Errorf("record at offset %d: %w", offset, err)
		}
		if len(encoded) > maxBatchBytes {
			return fmt.Errorf("encoded record at offset %d exceeds batch limit", offset)
		}

		if current != nil &&
			(current.count == maxBatchEvents ||
				current.data.Len()+len(encoded) > maxBatchBytes) {
			if err := flush(); err != nil {
				return err
			}
		}
		if current == nil {
			current = &batch{start: offset}
		}

		current.data.Write(encoded)
		current.count++
		offset += int64(len(line))
		current.end = offset
	}
}

func transform(line []byte, offset int64) ([]byte, error) {
	var in event
	decoder := json.NewDecoder(bytes.NewReader(line))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&in); err != nil {
		return nil, err
	}
	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
		return nil, fmt.Errorf("record contains trailing JSON")
	}
	if in.ID == "" || in.Account == "" {
		return nil, fmt.Errorf("id and account are required")
	}

	class := "zero"
	if in.AmountCents > 0 {
		class = "credit"
	} else if in.AmountCents < 0 {
		class = "debit"
	}

	encoded, err := json.Marshal(output{
		ID: in.ID, Account: in.Account, AmountCents: in.AmountCents,
		Class: class, SourceOffset: offset,
	})
	if err != nil {
		return nil, err
	}
	return append(encoded, '\n'), nil
}

func commitBatch(b *batch, cfg config, cp *checkpoint, logger *slog.Logger) error {
	if err := writeSegment(cfg.out, b); err != nil {
		return err
	}
	if cfg.failAfterSegment {
		return fmt.Errorf("injected failure after durable segment")
	}

	next := *cp
	next.Offset = b.end
	if err := saveCheckpoint(cfg.state, next); err != nil {
		return err
	}
	*cp = next
	logger.Info("batch_committed",
		"start_offset", b.start, "end_offset", b.end,
		"records", b.count, "bytes", b.data.Len())
	return nil
}

func writeSegment(outDir string, b *batch) error {
	final := filepath.Join(outDir,
		fmt.Sprintf("%020d-%020d.jsonl", b.start, b.end))

	tmp, err := os.CreateTemp(outDir, ".batch-*")
	if err != nil {
		return err
	}
	tmpName := tmp.Name()
	defer os.Remove(tmpName)

	if err := tmp.Chmod(0640); err != nil {
		tmp.Close()
		return err
	}
	n, err := tmp.Write(b.data.Bytes())
	if err != nil || n != b.data.Len() {
		tmp.Close()
		return fmt.Errorf("write temporary segment: %w", err)
	}
	if err := tmp.Sync(); err != nil {
		tmp.Close()
		return err
	}
	if err := tmp.Close(); err != nil {
		return err
	}

	if err := os.Link(tmpName, final); err != nil {
		if !errors.Is(err, os.ErrExist) {
			return fmt.Errorf("publish segment: %w", err)
		}
		same, compareErr := sameContent(final, b.data.Bytes())
		if compareErr != nil {
			return compareErr
		}
		if !same {
			return fmt.Errorf("existing segment %s has different content", final)
		}
	}
	if err := os.Remove(tmpName); err != nil && !errors.Is(err, os.ErrNotExist) {
		return err
	}
	return syncDir(outDir)
}

func sameContent(path string, expected []byte) (bool, error) {
	f, err := os.Open(path)
	if err != nil {
		return false, err
	}
	defer f.Close()

	info, err := f.Stat()
	if err != nil || !info.Mode().IsRegular() {
		return false, fmt.Errorf("existing segment is not a regular file")
	}
	if info.Size() != int64(len(expected)) {
		return false, nil
	}
	actualHash := sha256.New()
	if _, err := io.Copy(actualHash, f); err != nil {
		return false, err
	}
	expectedHash := sha256.Sum256(expected)
	return bytes.Equal(actualHash.Sum(nil), expectedHash[:]), nil
}

func loadCheckpoint(path string) (checkpoint, bool, error) {
	f, err := os.Open(path)
	if errors.Is(err, os.ErrNotExist) {
		return checkpoint{}, false, nil
	}
	if err != nil {
		return checkpoint{}, false, err
	}
	defer f.Close()

	var cp checkpoint
	decoder := json.NewDecoder(io.LimitReader(f, 64<<10))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&cp); err != nil {
		return cp, false, fmt.Errorf("decode checkpoint: %w", err)
	}
	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
		return cp, false, fmt.Errorf("checkpoint contains trailing data")
	}
	return cp, true, nil
}

func saveCheckpoint(path string, cp checkpoint) error {
	data, err := json.Marshal(cp)
	if err != nil {
		return err
	}
	data = append(data, '\n')
	dir := filepath.Dir(path)

	tmp, err := os.CreateTemp(dir, ".checkpoint-*")
	if err != nil {
		return err
	}
	tmpName := tmp.Name()
	defer os.Remove(tmpName)

	if err := tmp.Chmod(0600); err != nil {
		tmp.Close()
		return err
	}
	if _, err := tmp.Write(data); err != nil {
		tmp.Close()
		return err
	}
	if err := tmp.Sync(); err != nil {
		tmp.Close()
		return err
	}
	if err := tmp.Close(); err != nil {
		return err
	}
	if err := os.Rename(tmpName, path); err != nil {
		return err
	}
	return syncDir(dir)
}

func syncDir(path string) error {
	dir, err := os.Open(path)
	if err != nil {
		return err
	}
	defer dir.Close()
	return dir.Sync()
}

The checkpoint advances only after the corresponding segment is durable. On SIGTERM, the loop rechecks cancellation immediately after the read, refuses to transform the newly read record, and commits only the batch that was already in progress.

Build and test the happy path

mkdir -p sample/run/out
cat > sample/events.jsonl <<'EOF'
{"id":"evt-001","account":"alpha","amount_cents":1250}
{"id":"evt-002","account":"alpha","amount_cents":-300}
{"id":"evt-003","account":"beta","amount_cents":0}
EOF

go test ./...
go build -trimpath -o jsonstream ./cmd/jsonstream
./jsonstream \
  -input sample/events.jsonl \
  -out sample/run/out \
  -state sample/run/checkpoint.json

cat sample/run/out/*.jsonl
cat sample/run/checkpoint.json

The output should contain three classified records. Running the same command again reads EOF at the checkpoint and creates no additional segment.

Exercise failure recovery

The test flag injects a failure after publishing the segment but before updating the checkpoint. It models the durable state left by a crash at the most important boundary.

mkdir -p sample/recovery/out

if ./jsonstream \
  -input sample/events.jsonl \
  -out sample/recovery/out \
  -state sample/recovery/checkpoint.json \
  -fail-after-segment
then
  echo "expected injected failure" >&2
  exit 1
fi

segment=$(find sample/recovery/out -maxdepth 1 -type f -name '*.jsonl')
before=$(sha256sum "$segment" | cut -d' ' -f1)

./jsonstream \
  -input sample/events.jsonl \
  -out sample/recovery/out \
  -state sample/recovery/checkpoint.json

after=$(sha256sum "$segment" | cut -d' ' -f1)
test "$before" = "$after"
test "$(find sample/recovery/out -maxdepth 1 -type f -name '*.jsonl' | wc -l)" -eq 1
cat sample/recovery/checkpoint.json

Recovery encounters the existing segment, verifies its size and hash, and then advances the checkpoint. A mismatching segment stops processing instead of silently overwriting evidence.

Deploy as a hardened Linux service

Create deploy/jsonstream.service with the following configuration:

[Unit]
Description=Bounded JSON stream processor
After=local-fs.target
StartLimitIntervalSec=300
StartLimitBurst=3

[Service]
Type=exec
DynamicUser=yes
StateDirectory=jsonstream
StateDirectoryMode=0750
ExecStart=/usr/local/libexec/jsonstream \
  -input /srv/jsonstream/events.jsonl \
  -out /var/lib/jsonstream/out \
  -state /var/lib/jsonstream/checkpoint.json
Restart=on-failure
RestartSec=5s
TimeoutStopSec=30s
UMask=0027
Environment=GOMEMLIMIT=96MiB
MemoryMax=128M

NoNewPrivileges=yes
PrivateTmp=yes
PrivateDevices=yes
ProtectSystem=strict
ProtectHome=yes
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectControlGroups=yes
RestrictSUIDSGID=yes
LockPersonality=yes
CapabilityBoundingSet=
RestrictAddressFamilies=AF_UNIX
ReadOnlyPaths=/srv/jsonstream/events.jsonl

StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

These are host commands and require administrative privileges. Stage the real source before starting the service; do not modify it in place afterward.

sudo install -d -o root -g root -m 0755 /usr/local/libexec
sudo install -d -o root -g root -m 0755 /srv/jsonstream
sudo install -o root -g root -m 0755 jsonstream /usr/local/libexec/jsonstream
sudo install -o root -g root -m 0644 sample/events.jsonl /srv/jsonstream/events.jsonl
sudo install -o root -g root -m 0644 \
  deploy/jsonstream.service /etc/systemd/system/jsonstream.service

sudo systemctl daemon-reload
sudo systemctl enable --now jsonstream.service
sudo systemctl status jsonstream.service
sudo journalctl -u jsonstream.service --no-pager

The service opens no network sockets, so no firewall rule is necessary. Keeping network access out of this worker also narrows its failure and security surface. The dynamic user can write only beneath the systemd-managed state directory.

Observability and performance

Each committed batch logs offsets, record count, and output bytes as structured JSON. The completion event reports the final offset and source size. Operational monitoring should alert on repeated stream_failed events and measure checkpoint lag as source_size - offset.

Batch size is the central durability-throughput trade-off. Larger batches amortize file creation, hashing, checkpoint replacement, and fsync, but increase replay work and visibility latency. Smaller batches do the reverse. Measure with representative record sizes and storage, including forced restarts.

The 4 MiB batch and 1 MiB record limits bound application payload buffering, not the entire Go process. Decoder strings, runtime metadata, filesystem buffers, and garbage collection require additional memory. GOMEMLIMIT guides the runtime, while systemd’s MemoryMax supplies the hard containment boundary.

Common production failures

• Source hash mismatch: the input changed or the wrong checkpoint was selected. Start a new job with a new state and output directory.
• Poison record: malformed JSON, unknown fields, or missing identifiers stop at a stable byte offset. Validate data before staging; do not skip records invisibly.
• Segment mismatch: code changed, output was edited, or directories were reused across jobs. Preserve the files for investigation.
• Permission or synchronization failure: confirm ownership, mount options, free space, and inode availability. Treat failed durability calls as failed batches.
• Restart loop: deterministic input errors will recur. The systemd start limit prevents an unbounded tight loop, but operators still need an alert.
• Forced shutdown: SIGKILL can interrupt persistence, but restart reconciliation remains safe. Keep the shutdown budget longer than observed worst-case synchronization time.

Final verification checklist

• The source is immutable, newline-terminated, and stored separately from output.
• Memory limits account for runtime overhead as well as batch constants.
• A normal rerun creates no duplicate segment.
• The injected post-segment failure recovers without changing segment content.
• SIGTERM stops new transformations and commits only prior in-flight work.
• Logs expose committed offsets, failures, and completion.
• Downstream effects use an idempotency key and assume at-least-once delivery.
• The deployed filesystem’s link, rename, and directory-sync behavior has been tested.

Reliable stream processing is less about an impressive loop and more about choosing one precise boundary: output first, checkpoint second, with deterministic recovery between them. Once that boundary is durable, bounded, observable, and testable, crashes stop being mysterious events. They become another input the system already knows how to process.