build-binaries / build-binaries (push) Successful in 2m21s
Closes #79. Emit is latency-bound, not CPU-bound. Every blob costs two serial round-trips to the zone — a `ProbeKey` HEAD, then a `PutReader` PUT — so a hak with a few thousand resources pays a few thousand serialised latencies. Measured on the live sow-assets-manifest backfill: 26 s of CPU across 9.5 minutes of wall clock, on a 4-core host with 5 GB free and peak RSS of 51 MB. `emit` now hashes, compresses and stores `--jobs N` resources at once, default 16 — matching `DEPOT_JOBS` and the transport's `MaxIdleConnsPerHost`, so a worker per connection needs no fresh TLS handshake. `--jobs 1` is exactly the old behaviour. ### Three properties had to survive Each has a test in `internal/nwsync/jobs_test.go`: - **Deterministic manifest bytes.** `emitterVersion` promises a manifest is a function of its artifact, 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 uploads finish in. `TestEmitProducesTheSameIndexAtEveryJobCount` diffs the `.nsym` and its sidecar between `-jobs 1` and `-jobs 16`. - **Index still lands last.** Any worker's failure aborts before a manifest is written. `TestEmitLeavesNoIndexWhenAParallelUploadFails` fails every blob PUT with 16 workers in flight and asserts no `.nsym` appears. Under `-race` it also covers the shared counters. - **Identical content still shares one blob.** This one bit during development and is the reason to read the diff carefully: serially, the sink's existence check absorbed two resrefs with identical bytes. In parallel both workers probe, both miss, and both upload — `TestEmitWritesBlobsAndManifest` caught it as "wrote 2 blobs, want 1". Claiming the sha1 in-process restores the dedupe and skips a probe round-trip as well. ### Memory Peak now tracks the resources in flight rather than one resource. The ceiling is `N` × the 15 MB `fileSizeLimit` plus its compressed copy — bounded by a constant this package enforces itself, and still not tracking the archive. `TestEmitPeakMemoryIsBoundedByJobCount` re-runs the #76 regression check at `-jobs 8`: a hak 8× bigger still costs the same. This is only cheap because of #78. Before streaming emit, N workers would have meant N whole archives resident. ### Not done Skipping the `ProbeKey` HEAD on a first-time emit would halve round-trips, but doubles uploaded bytes on a re-run — which is exactly what a backfill is. Noted in #79 so it is not rediscovered; parallelism is the better lever and this PR takes it. Worker compression still serialises on `blobEncoder`, which is `WithEncoderConcurrency(1)` for the memory reason in `compressedbuf.go`. At 26 s of CPU per hak that is not worth trading memory for, but it is where to look if the numbers ever say otherwise. 🤖 Generated with [Claude Code](https://claude.com/claude-code)Reviewed-on: #80 Co-authored-by: vickydotbat <vickydotbat@tutamail.com>
147 lines
4.7 KiB
Go
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
|
|
}
|