Files
sow-tools/internal/nwsync/run.go
T
archvillainetteandClaude Opus 5 f69121a9e8
ci / ci (pull_request) Successful in 3m32s
feat(nwsync): emit blobs in parallel with a bounded worker pool (#79)
Emit is latency-bound, not CPU-bound. Every blob costs two serial HTTP
round-trips to the zone — an existence probe, then an upload — so a hak
with a few thousand resources pays a few thousand serialised latencies.
A measured backfill spent 26 seconds of CPU across 9.5 minutes of wall
clock, on a host with three of four cores idle and 5 GB free.

Emit now hashes, compresses and stores `--jobs N` resources at once
(default 16, matching DEPOT_JOBS and the transport's idle connections
per host).

Three properties had to survive, and each has a test:

- The manifest's bytes are promised deterministic by emitterVersion, so
  entries is index-addressed rather than appended to: a worker owns
  entries[i] alone and the slice comes back in artifact order whatever
  order the uploads finish in.
- The index is still the publication marker, so any worker's failure
  aborts the run before a manifest is written. Workers drain the rest of
  the channel instead of returning, which keeps the feeder from blocking
  on workers that have gone away.
- Two resrefs holding identical bytes still share one blob. Serially the
  sink's existence check absorbed that; in parallel both workers would
  probe, both miss, and both upload. Claiming the sha1 in-process
  restores the dedupe and skips a probe round-trip as well.

Peak memory is now the resources in flight rather than one resource, so
the ceiling is N times the 15 MB per-resource limit and its compressed
copy — bounded by a constant this package enforces itself, and still
nowhere near tracking the archive.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-31 16:28:45 +02:00

147 lines
4.7 KiB
Go

// Package nwsync publishes NWSync repository data: blobs and a per-artifact
// NSYM manifest at each artifact's birth (emit), and one merged manifest at
// module release (assemble).
//
// The split exists because upstream nwn_nwsync_write wants every hak, the TLK
// and the module present in one run on one disk, which our build hosts cannot
// hold. Upstream stays the conformance oracle: manifests compare byte for
// byte, blobs compare after decompression.
package nwsync
import (
"flag"
"fmt"
"io"
)
const (
exitOK = 0
exitUsage = 64
exitInternal = 70
)
// Run executes an nwsync subcommand. args[0] is the subcommand (emit|assemble);
// returns the process exit code.
func Run(args []string, stdout, stderr io.Writer) int {
if len(args) == 0 {
printRunUsage(stderr)
return exitUsage
}
switch args[0] {
case "emit":
return runEmit(args[1:], stdout, stderr)
case "assemble":
return runAssemble(args[1:], stdout, stderr)
case "-h", "--help", "help":
printRunUsage(stdout)
return exitOK
default:
fmt.Fprintf(stderr, "nwsync: unknown subcommand %q\n\n", args[0])
printRunUsage(stderr)
return exitUsage
}
}
func printRunUsage(w io.Writer) {
fmt.Fprint(w, `usage:
nwsync emit [--as NAME] [--out DIR] <artifact-key> <file>
nwsync assemble --group-id N [--tlk-key KEY] [--out DIR] <artifact-key>...
emit explodes one .hak/.erf or one loose file (the TLK) into NWSync blobs plus
a NSYM index covering only that artifact, and uploads both. assemble merges
those indexes into one manifest, reading no bulk data. Artifact keys are depot
keys; an index lives beside its artifact, with the extension replaced.
--out DIR writes to a local repository tree instead of uploading, which is the
conformance path against upstream nwn_nwsync_write. Without it, the zone comes
from NWSYNC_STORAGE_ZONE, NWSYNC_STORAGE_PASSWORD and BUNNY_STORAGE_HOST.
`)
}
// parseArgs parses flags that may appear before, after or between positionals.
// Go's flag package stops at the first non-flag argument, which turns
// `emit <key> <file> --out DIR` into a confusing arity error.
func parseArgs(fs *flag.FlagSet, args []string) ([]string, error) {
var positional []string
for {
if err := fs.Parse(args); err != nil {
return nil, err
}
rest := fs.Args()
if len(rest) == 0 {
return positional, nil
}
positional = append(positional, rest[0])
args = rest[1:]
}
}
func runEmit(args []string, stdout, stderr io.Writer) int {
fs := flag.NewFlagSet("emit", flag.ContinueOnError)
fs.SetOutput(stderr)
as := fs.String("as", "", "published name of the artifact, when it differs from the key")
out := fs.String("out", "", "write to a local repository tree instead of uploading")
jobs := fs.Int("jobs", defaultEmitJobs, "resources to hash, compress and store at once")
positional, err := parseArgs(fs, args)
if err != nil {
return exitUsage
}
if len(positional) != 2 {
fmt.Fprintf(stderr, "nwsync emit: <artifact-key> and <file> are both required\n")
return exitUsage
}
if *jobs < 1 {
fmt.Fprintf(stderr, "nwsync emit: -jobs must be at least 1, got %d\n", *jobs)
return exitUsage
}
result, err := Emit(EmitOptions{
ArtifactKey: positional[0],
ArtifactPath: positional[1],
As: *as,
OutDir: *out,
Jobs: *jobs,
})
if err != nil {
fmt.Fprintf(stderr, "nwsync emit: %v\n", err)
return exitInternal
}
fmt.Fprintf(stdout, "emitted %s: %d resources, %d new blobs, index %s\n",
result.Name, result.Entries, result.BlobsWritten, result.ManifestPath)
return exitOK
}
func runAssemble(args []string, stdout, stderr io.Writer) int {
fs := flag.NewFlagSet("assemble", flag.ContinueOnError)
fs.SetOutput(stderr)
tlkKey := fs.String("tlk-key", "", "depot key of the TLK, which shadows nothing and merges last")
out := fs.String("out", "", "write to a local repository tree instead of uploading")
groupID := fs.Int("group-id", 0, "NWSync group id (1 = current, 2 = testing; 0 omits it)")
moduleName := fs.String("module-name", "", "module name recorded in the sidecar")
description := fs.String("description", "", "description recorded in the sidecar")
positional, err := parseArgs(fs, args)
if err != nil {
return exitUsage
}
if len(positional) == 0 {
fmt.Fprintf(stderr, "nwsync assemble: at least one artifact key is required\n")
return exitUsage
}
result, err := Assemble(AssembleOptions{
ArtifactKeys: positional,
TLKKey: *tlkKey,
OutDir: *out,
GroupID: *groupID,
ModuleName: *moduleName,
Description: *description,
})
if err != nil {
fmt.Fprintf(stderr, "nwsync assemble: %v\n", err)
return exitInternal
}
fmt.Fprintf(stdout, "assembled manifest %s: %d resources, %s\n",
result.SHA1, result.Entries, result.ManifestPath)
return exitOK
}