nwsync: upload sink, key-addressed CLI, fail-closed publication marker (#73)

Builds the code half of #60, and closes the CLI gap the #53 sweep found: PR #71 shipped #53's original surface rather than the one #56, #62 and #65 settled.

## What lands

**`depot.KeyStore`** — `ProbeKey` / `PutReader` / `GetKey`, addressing the zone by object key rather than by depot sha. NWSync cannot use the sha-addressed path: a blob is named after the sha1 of its *uncompressed* bytes while the body uploaded is the compressed form, and Bunny's `Checksum` header is sha256 of the body. Per #55 this reuses `internal/depot`'s `httpBackend` — same IPv4-pinned transport, same retry, same tri-state probe — and the sha-addressed `Backend` is now rewritten on top of it. No second HTTP client, no per-instance hash-function fields: the caller passes the key and the checksum, which turned out simpler than #55 expected.

**A sink in `internal/nwsync`** — the zone by default, a local tree under `--out DIR` as the conformance path. Blobs upload as they are produced and the index lands last, so the presence of an index is the publication marker. A blob already in the zone is skipped via #55's probe *without* paying for compression (the body is a thunk) — which matters for the backfill, where compression is the expensive part.

**The settled CLI**

```
nwsync emit     [--as NAME] [--out DIR] <artifact-key> <file>
nwsync assemble --group-id N [--tlk-key KEY] [--out DIR] <artifact-key>...
```

Artifact keys are depot keys; an index lives beside its artifact with the extension replaced (#62), derived in exactly one place so `emit` and `assemble` cannot disagree. Flags may now follow positionals — Go's `flag` stops at the first non-flag argument, which cost a run during #59.

**Fail-closed in two places** — an artifact key whose embedded digest does not match the file is refused (publishing an index under the wrong key silently pairs a manifest with the wrong artifact), and `assemble` refuses an artifact with no index rather than publishing a manifest missing a hak.

## Checks

`make check` green. Six new tests run against a Bunny-shaped `httptest` zone that verifies the `Checksum` header the way Bunny does: blobs-then-index ordering, skip-if-present, no index after a failed upload, key/file mismatch, and assemble reading indexes back out of the zone.

Conformance re-run through the new CLI against upstream `nwn_nwsync_write` 2.1.2 over `sow_vfxs_01.hak` (#59's oracle): the manifest is still **byte-identical**.

## Not in this PR

- **Live upload against the real zone.** The nwsync zone and its credential are #61, still open. Everything here is proven against a fake zone only.
- **Consumer wiring** — #65, in the three producer repos.
- **The mid-hak failure *policy*.** The mechanism is here (fail closed, orphan blobs left, re-run resumes); whether a module release may proceed when an emit failed is a human call, still open on #60.

🤖 Generated with [Claude Code](https://claude.com/claude-code)Reviewed-on: #73

Co-authored-by: vickydotbat <vickydotbat@tutamail.com>
This commit is contained in:
2026-07-30 21:56:06 +00:00
committed by archvillainette
parent a131b25e5b
commit 667e7648bf
9 changed files with 841 additions and 203 deletions
+26 -4
View File
@@ -43,10 +43,32 @@ flags are mutually exclusive.
`nwsync emit` runs where an artifact is born (a `.hak`/`.erf`, or a loose file `nwsync emit` runs where an artifact is born (a `.hak`/`.erf`, or a loose file
such as the TLK); `nwsync assemble` runs at module release and reads only the such as the TLK); `nwsync assemble` runs at module release and reads only the
small per-artifact manifests. `--order` lists artifact names highest priority small per-artifact indexes. Both take **depot keys**: an artifact's index lives
first: a resref in more than one artifact resolves to the earliest one, the way beside the artifact itself with the extension replaced, so `emit` and
the game resolves it. `--group-id` is per channel — 1 is current, 2 is testing, `assemble` agree on where it is without being told.
and 0 leaves the field out of the sidecar.
```
nwsync emit [--as NAME] [--out DIR] <artifact-key> <file>
nwsync assemble --group-id N [--tlk-key KEY] [--out DIR] <artifact-key>...
```
Both verbs upload by default; nothing bulky is ever written to the runner's
disk. `--out DIR` writes a local repository tree instead, which is the
conformance path against upstream `nwn_nwsync_write`. The zone comes from
`NWSYNC_STORAGE_ZONE` and `NWSYNC_STORAGE_PASSWORD`, with the host from
`BUNNY_STORAGE_HOST` — NWSync data is a separate zone from the asset depot.
`assemble`'s artifact keys are in `Mod_HakList` order, highest priority first: a
resref in more than one artifact resolves to the earliest one, the way the game
resolves it. `--tlk-key` has its own slot because the TLK shadows nothing.
`--group-id` is per channel — 1 is current, 2 is testing, and 0 leaves the field
out of the sidecar.
`emit` uploads blobs first and the index last, so the presence of an index is
the publication marker: an artifact whose emit died halfway leaves real blobs in
the zone and no index. Blob names are content hashes, so re-running skips
whatever already landed, and `assemble` fails closed on an artifact with no
index rather than publishing a manifest that is missing a hak.
## Hidden compatibility aliases ## Hidden compatibility aliases
+121
View File
@@ -0,0 +1,121 @@
package depot
import (
"context"
"errors"
"fmt"
"io"
"net/http"
"os"
"strings"
)
// KeyStore is the zone addressed by object key rather than by depot sha. The
// depot names every object after the sha256 of its contents; NWSync does not —
// a blob is named after the sha1 of its *uncompressed* bytes while the body
// uploaded is the compressed form, and a per-artifact index is named after its
// artifact. Both addressing modes want the same transport, retry and probe
// discipline, so the sha-addressed Backend rides on this rather than the other
// way round.
type KeyStore interface {
// ProbeKey returns the existence state of one key. transient=true means a
// retry might change the answer — never read it as "missing, re-upload".
ProbeKey(ctx context.Context, key string) (state ProbeState, transient bool, err error)
// PutReader uploads size bytes read from r to key. checksum is the
// uppercase hex sha256 of those bytes, which Bunny verifies server-side.
PutReader(ctx context.Context, key string, r io.Reader, size int64, checksum string) error
// GetKey fetches the whole object at key. Small objects only — it holds
// the body in memory and does no hash check, because a key is not always
// a content hash.
GetKey(ctx context.Context, key string) ([]byte, error)
}
// NewKeyStore returns a KeyStore for cfg's storage zone. Fails closed on a
// missing host or read key, matching NewBackend.
func NewKeyStore(cfg Config) (KeyStore, error) {
if cfg.StorageHost == "" {
return nil, errors.New("storage backend requires a storage host")
}
if cfg.StorageZone == "" {
return nil, errors.New("storage backend requires a storage zone")
}
if cfg.ReadKey == "" {
return nil, errors.New("storage backend requires a read key")
}
return &httpBackend{name: "bunny", client: newHTTPClient(cfg), cfg: cfg}, nil
}
// keyURL is the storage URL of one object key.
func (b *httpBackend) keyURL(key string) string {
host := b.cfg.StorageHost
// StorageHost is normally a bare host ("storage.bunnycdn.com"); allow a
// full scheme (used by tests against httptest.NewServer) to pass through
// unchanged.
if !strings.Contains(host, "://") {
host = "https://" + host
}
return fmt.Sprintf("%s/%s/%s", strings.TrimSuffix(host, "/"), b.cfg.StorageZone, key)
}
func (b *httpBackend) ProbeKey(ctx context.Context, key string) (ProbeState, bool, error) {
return b.rangeProbe(ctx, b.keyURL(key), map[string]string{"AccessKey": b.cfg.ReadKey})
}
func (b *httpBackend) PutReader(ctx context.Context, key string, r io.Reader, size int64, checksum string) error {
if b.name == "cdn" {
return errors.New("cdn backend is read-only")
}
if b.cfg.WriteKey == "" {
return errors.New("storage backend requires a write key to write")
}
req, err := http.NewRequestWithContext(ctx, http.MethodPut, b.keyURL(key), r)
if err != nil {
return err
}
req.ContentLength = size
req.Header.Set("AccessKey", b.cfg.WriteKey)
// Bunny defines Checksum as sha256 of the body and rejects a mismatch, so
// this is server-side integrity checking, not decoration.
req.Header.Set("Checksum", strings.ToUpper(checksum))
resp, err := b.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
_, _ = io.Copy(io.Discard, resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("put %s: unexpected status %d", key, resp.StatusCode)
}
return nil
}
func (b *httpBackend) GetKey(ctx context.Context, key string) ([]byte, error) {
resp, err := b.get(ctx, b.keyURL(key), map[string]string{"AccessKey": b.cfg.ReadKey})
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
_, _ = io.Copy(io.Discard, resp.Body)
return nil, fmt.Errorf("get %s: unexpected status %d", key, resp.StatusCode)
}
return io.ReadAll(resp.Body)
}
// putFile uploads the file at src to key, streaming it. checksum is the
// uppercase hex sha256 of the file's bytes.
func (b *httpBackend) putFile(ctx context.Context, key, src, checksum string) error {
f, err := os.Open(src)
if err != nil {
return err
}
defer f.Close()
info, err := f.Stat()
if err != nil {
return err
}
return b.PutReader(ctx, key, f, info.Size(), checksum)
}
+4 -36
View File
@@ -10,7 +10,6 @@ import (
"net/http" "net/http"
"os" "os"
"path/filepath" "path/filepath"
"strings"
"time" "time"
) )
@@ -71,16 +70,7 @@ type httpBackend struct {
func (b *httpBackend) Name() string { return b.name } func (b *httpBackend) Name() string { return b.name }
func (b *httpBackend) storageURL(sha string) string { func (b *httpBackend) storageURL(sha string) string { return b.keyURL(BlobKey(sha)) }
host := b.cfg.StorageHost
// StorageHost is normally a bare host ("storage.bunnycdn.com"); allow a
// full scheme (used by tests against httptest.NewServer) to pass through
// unchanged.
if strings.Contains(host, "://") {
return fmt.Sprintf("%s/%s/%s", strings.TrimSuffix(host, "/"), b.cfg.StorageZone, BlobKey(sha))
}
return fmt.Sprintf("https://%s/%s/%s", host, b.cfg.StorageZone, BlobKey(sha))
}
func (b *httpBackend) cdnURL(sha string) string { func (b *httpBackend) cdnURL(sha string) string {
return fmt.Sprintf("%s/%s", b.cfg.CDNBase, BlobKey(sha)) return fmt.Sprintf("%s/%s", b.cfg.CDNBase, BlobKey(sha))
@@ -140,7 +130,8 @@ func (b *httpBackend) Probe(ctx context.Context, sha string) (ProbeState, bool,
return b.rangeProbe(ctx, b.storageURL(sha), map[string]string{"AccessKey": b.cfg.ReadKey}) return b.rangeProbe(ctx, b.storageURL(sha), map[string]string{"AccessKey": b.cfg.ReadKey})
} }
// Put uploads src for sha. cdn is read-only. // Put uploads src for sha. cdn is read-only. A depot object is named after the
// sha256 of its own bytes, so the key's sha doubles as the Checksum header.
func (b *httpBackend) Put(ctx context.Context, sha, src string) error { func (b *httpBackend) Put(ctx context.Context, sha, src string) error {
if b.name == "cdn" { if b.name == "cdn" {
return errors.New("cdn backend is read-only") return errors.New("cdn backend is read-only")
@@ -157,30 +148,7 @@ func (b *httpBackend) Put(ctx context.Context, sha, src string) error {
return nil return nil
} }
f, err := os.Open(src) return b.putFile(ctx, BlobKey(sha), src, sha)
if err != nil {
return err
}
defer f.Close()
req, err := http.NewRequestWithContext(ctx, http.MethodPut, b.storageURL(sha), f)
if err != nil {
return err
}
req.Header.Set("AccessKey", b.cfg.WriteKey)
req.Header.Set("Checksum", strings.ToUpper(sha))
resp, err := b.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
_, _ = io.Copy(io.Discard, resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return fmt.Errorf("bunny put %s: unexpected status %d", sha, resp.StatusCode)
}
return nil
} }
// Get fetches sha into dest via temp file + rename, re-hashing and deleting // Get fetches sha into dest via temp file + rename, re-hashing and deleting
+40 -23
View File
@@ -4,18 +4,21 @@ import (
"crypto/sha1" "crypto/sha1"
"encoding/json" "encoding/json"
"fmt" "fmt"
"os" "path"
"path/filepath"
) )
// AssembleOptions describes one merged manifest. // AssembleOptions describes one merged manifest.
type AssembleOptions struct { type AssembleOptions struct {
Order []string // artifact names, highest priority first // ArtifactKeys are the depot keys of the artifacts to merge, in
EntriesDir string // directory holding <name>.nsym and <name>.nsym.json // Mod_HakList order — highest priority first. Each one's index is read
OutDir string // repository root; the manifest lands in <out>/manifests // from the key beside it.
ArtifactKeys []string
TLKKey string // the TLK's key, if the manifest carries one
OutDir string // write locally instead of uploading — the conformance path
GroupID int // 1 = current, 2 = testing; 0 means absent GroupID int // 1 = current, 2 = testing; 0 means absent
ModuleName string ModuleName string
Description string Description string
Sink sink // test seam; nil means OutDir or the zone
} }
// AssembleResult reports what one assemble run produced. // AssembleResult reports what one assemble run produced.
@@ -33,25 +36,43 @@ type AssembleResult struct {
// the game resolves it (upstream's resman adds haks in reverse and lets the // the game resolves it (upstream's resman adds haks in reverse and lets the
// last one win). Get this backwards and the wrong texture ships silently. // last one win). Get this backwards and the wrong texture ships silently.
func Assemble(options AssembleOptions) (AssembleResult, error) { func Assemble(options AssembleOptions) (AssembleResult, error) {
if len(options.Order) == 0 { if len(options.ArtifactKeys) == 0 {
return AssembleResult{}, fmt.Errorf("assemble: --order names no artifacts") return AssembleResult{}, fmt.Errorf("assemble: no artifact keys given")
}
target, err := openSink(options.OutDir, options.Sink)
if err != nil {
return AssembleResult{}, err
}
// The TLK carries no precedence — it is not a hak and shadows nothing —
// so it merges last, after every hak has had its say.
keys := append([]string{}, options.ArtifactKeys...)
if options.TLKKey != "" {
keys = append(keys, options.TLKKey)
} }
merged := make([]Entry, 0, 1024) merged := make([]Entry, 0, 1024)
winner := make(map[Identity]bool, 1024) winner := make(map[Identity]bool, 1024)
var onDiskBytes int64 var onDiskBytes int64
for _, name := range options.Order { for _, artifactKey := range keys {
manifestPath := filepath.Join(options.EntriesDir, name+".nsym") key, err := resolveIndexKey(artifactKey, options.OutDir)
data, err := os.ReadFile(manifestPath)
if err != nil { if err != nil {
return AssembleResult{}, fmt.Errorf("assemble: no index for %q: %w", name, err) return AssembleResult{}, err
}
data, sidecarBody, err := target.getIndex(key)
if err != nil {
// An artifact with no index is an artifact whose emit never
// finished. Publishing a manifest without it would ship a release
// missing a hak, so this fails closed.
return AssembleResult{}, fmt.Errorf("assemble: no index for %s: %w", artifactKey, err)
} }
entries, err := readManifest(data) entries, err := readManifest(data)
if err != nil { if err != nil {
return AssembleResult{}, fmt.Errorf("%s: %w", manifestPath, err) return AssembleResult{}, fmt.Errorf("%s: %w", target.describe(key), err)
} }
sidecar, err := readSidecar(manifestPath + ".json") sidecar, err := parseSidecar(target.describe(key), sidecarBody)
if err != nil { if err != nil {
return AssembleResult{}, err return AssembleResult{}, err
} }
@@ -61,7 +82,7 @@ func Assemble(options AssembleOptions) (AssembleResult, error) {
if sidecar.EmitterVersion != emitterVersion { if sidecar.EmitterVersion != emitterVersion {
return AssembleResult{}, fmt.Errorf( return AssembleResult{}, fmt.Errorf(
"assemble: emitter version mismatch: %s was emitted by emitter %q, this is emitter %q", "assemble: emitter version mismatch: %s was emitted by emitter %q, this is emitter %q",
name, sidecar.EmitterVersion, emitterVersion) artifactKey, sidecar.EmitterVersion, emitterVersion)
} }
// on_disk_bytes overcounts by the handful of cross-artifact // on_disk_bytes overcounts by the handful of cross-artifact
// duplicates. It is a display statistic; no dedupe pass for it. // duplicates. It is a display statistic; no dedupe pass for it.
@@ -86,26 +107,22 @@ func Assemble(options AssembleOptions) (AssembleResult, error) {
return AssembleResult{}, err return AssembleResult{}, err
} }
sha1Hex := fmt.Sprintf("%x", sha1.Sum(data)) sha1Hex := fmt.Sprintf("%x", sha1.Sum(data))
manifestPath := filepath.Join(options.OutDir, "manifests", sha1Hex) manifestKey := path.Join("manifests", sha1Hex)
sidecar := Sidecar{ sidecar := Sidecar{
ModuleName: options.ModuleName, ModuleName: options.ModuleName,
Description: options.Description, Description: options.Description,
GroupID: options.GroupID, GroupID: options.GroupID,
} }
if err := writeManifestPair(manifestPath, data, merged, onDiskBytes, sidecar); err != nil { if err := putManifestPair(target, manifestKey, data, merged, onDiskBytes, sidecar); err != nil {
return AssembleResult{}, err return AssembleResult{}, err
} }
return AssembleResult{SHA1: sha1Hex, ManifestPath: manifestPath, Entries: len(merged)}, nil return AssembleResult{SHA1: sha1Hex, ManifestPath: target.describe(manifestKey), Entries: len(merged)}, nil
} }
func readSidecar(path string) (Sidecar, error) { func parseSidecar(where string, data []byte) (Sidecar, error) {
data, err := os.ReadFile(path)
if err != nil {
return Sidecar{}, fmt.Errorf("assemble: missing sidecar: %w", err)
}
var sidecar Sidecar var sidecar Sidecar
if err := json.Unmarshal(data, &sidecar); err != nil { if err := json.Unmarshal(data, &sidecar); err != nil {
return Sidecar{}, fmt.Errorf("%s: %w", path, err) return Sidecar{}, fmt.Errorf("%s: %w", where, err)
} }
return sidecar, nil return sidecar, nil
} }
+71 -54
View File
@@ -2,9 +2,11 @@ package nwsync
import ( import (
"bytes" "bytes"
"context"
"crypto/sha1" "crypto/sha1"
"fmt" "fmt"
"os" "os"
"path"
"path/filepath" "path/filepath"
"slices" "slices"
"sort" "sort"
@@ -60,42 +62,86 @@ type EmitResult struct {
BlobsWritten int BlobsWritten int
} }
// Emit explodes one artifact — a .hak/.erf/.mod or a loose file such as the // EmitOptions describes one emit run.
// TLK — into NWSync blobs plus a NSYM manifest describing only that artifact. type EmitOptions struct {
func Emit(artifactPath, outDir string) (EmitResult, error) { ArtifactKey string // depot key of the artifact; the NSYM key is derived from it
resources, err := readArtifact(artifactPath) ArtifactPath string // the file on disk
As string // name override, for a TLK whose filename is not its published name
OutDir string // write locally instead of uploading — the conformance path
Sink sink // test seam; nil means OutDir or the zone
}
// Emit explodes one artifact — a .hak/.erf or a loose file such as the TLK —
// into NWSync blobs plus a NSYM manifest describing only that artifact.
//
// Blobs go up as they are produced and the index lands last, so the presence of
// an index is the publication marker: an artifact whose emit died halfway has
// real blobs in the zone and no index, which is unambiguous. Blob names are
// content hashes, so re-running skips whatever already landed.
func Emit(options EmitOptions) (EmitResult, error) {
artifact, err := os.ReadFile(options.ArtifactPath)
if err != nil {
return EmitResult{}, fmt.Errorf("read artifact: %w", err)
}
if err := checkArtifactKey(options.ArtifactKey, artifact); err != nil {
return EmitResult{}, err
}
name := options.As
if name == "" {
name = path.Base(options.ArtifactKey)
}
extension := path.Ext(name)
name = strings.TrimSuffix(name, extension)
key, err := resolveIndexKey(options.ArtifactKey, options.OutDir)
if err != nil { if err != nil {
return EmitResult{}, err return EmitResult{}, err
} }
name := strings.TrimSuffix(filepath.Base(artifactPath), filepath.Ext(artifactPath))
entries, blobs, onDiskBytes, err := emitResources(resources, outDir) resources, err := readArtifact(options.ArtifactPath, artifact, name)
if err != nil {
return EmitResult{}, err
}
target, err := openSink(options.OutDir, options.Sink)
if err != nil {
return EmitResult{}, err
}
entries, blobs, onDiskBytes, err := emitResources(resources, target)
if err != nil { if err != nil {
return EmitResult{}, err return EmitResult{}, err
} }
if len(entries) == 0 { if len(entries) == 0 {
return EmitResult{}, fmt.Errorf("%s: nothing to index (no publishable resources)", artifactPath) return EmitResult{}, fmt.Errorf("%s: nothing to index (no publishable resources)", options.ArtifactPath)
} }
data, err := writeManifest(entries) data, err := writeManifest(entries)
if err != nil { if err != nil {
return EmitResult{}, err return EmitResult{}, err
} }
manifestPath := filepath.Join(outDir, name+".nsym") if err := putManifestPair(target, key, data, entries, onDiskBytes, Sidecar{ModuleName: name}); err != nil {
if err := writeManifestPair(manifestPath, data, entries, onDiskBytes, Sidecar{ModuleName: name}); err != nil {
return EmitResult{}, err return EmitResult{}, err
} }
return EmitResult{Name: name, ManifestPath: manifestPath, Entries: len(entries), BlobsWritten: blobs}, nil return EmitResult{Name: name, ManifestPath: target.describe(key), Entries: len(entries), BlobsWritten: blobs}, nil
}
// openSink returns the zone sink, or a local directory when outDir is set.
func openSink(outDir string, injected sink) (sink, error) {
if injected != nil {
return injected, nil
}
if outDir != "" {
return dirSink{root: outDir}, nil
}
return newZoneSink(context.Background(), os.Getenv)
} }
// readArtifact returns the resources of an ERF/HAK/MOD, or the single resource // readArtifact returns the resources of an ERF/HAK/MOD, or the single resource
// a loose file represents. Upstream's resman does the same dispatch on the // a loose file represents. Upstream's resman does the same dispatch on the
// file's first three bytes. // file's first three bytes. name is the artifact's published name, which for a
func readArtifact(path string) ([]erf.Resource, error) { // loose file is also its resref.
data, err := os.ReadFile(path) func readArtifact(path string, data []byte, name string) ([]erf.Resource, error) {
if err != nil {
return nil, fmt.Errorf("read artifact: %w", err)
}
if len(data) >= 3 { if len(data) >= 3 {
switch string(data[:3]) { switch string(data[:3]) {
case "ERF", "HAK": case "ERF", "HAK":
@@ -111,20 +157,19 @@ func readArtifact(path string) ([]erf.Resource, error) {
return nil, fmt.Errorf("%s: a module is never emitted; a manifest is haks plus the TLK", path) return nil, fmt.Errorf("%s: a module is never emitted; a manifest is haks plus the TLK", path)
} }
} }
base := filepath.Base(path) extension := filepath.Ext(filepath.Base(path))
extension := filepath.Ext(base)
restype, ok := erf.ResourceTypeForExtension(extension) restype, ok := erf.ResourceTypeForExtension(extension)
if !ok { if !ok {
return nil, fmt.Errorf("%s: unknown resource type %q", path, extension) return nil, fmt.Errorf("%s: unknown resource type %q", path, extension)
} }
return []erf.Resource{{ return []erf.Resource{{
Name: strings.TrimSuffix(base, extension), Name: name,
Type: restype, Type: restype,
Data: data, Data: data,
}}, nil }}, nil
} }
func emitResources(resources []erf.Resource, outDir string) ([]Entry, int, int64, error) { func emitResources(resources []erf.Resource, target sink) ([]Entry, int, int64, error) {
// A resref appearing twice inside one artifact resolves to the last one, // A resref appearing twice inside one artifact resolves to the last one,
// the way resman lets the last container added win. // the way resman lets the last container added win.
order := make([]Identity, 0, len(resources)) order := make([]Identity, 0, len(resources))
@@ -164,7 +209,7 @@ func emitResources(resources []erf.Resource, outDir string) ([]Entry, int, int64
ResRef: identity.ResRef, ResRef: identity.ResRef,
ResType: identity.ResType, ResType: identity.ResType,
}) })
written, err := writeBlob(outDir, sum, resource.Data) written, err := target.putBlob(fmt.Sprintf("%x", sum), func() []byte { return compressBlob(resource.Data) })
if err != nil { if err != nil {
return nil, 0, 0, err return nil, 0, 0, err
} }
@@ -187,35 +232,10 @@ func created() int64 {
return time.Now().Unix() return time.Now().Unix()
} }
// writeBlob writes one NWCompressedBuffer blob and returns its size on disk, // putManifestPair stores a NSYM manifest and its .json sidecar at key. The
// or 0 if the blob already existed. Blob names are content hashes, so an // caller supplies the sidecar fields it knows; the rest are derived from the
// existing name is existing content. // entries. data must be the serialised form of entries.
func writeBlob(outDir string, sum [20]byte, data []byte) (int64, error) { func putManifestPair(target sink, key string, data []byte, entries []Entry, onDiskBytes int64, sidecar Sidecar) error {
path := blobPath(outDir, fmt.Sprintf("%x", sum))
if _, err := os.Stat(path); err == nil {
return 0, nil
}
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return 0, fmt.Errorf("create blob directory: %w", err)
}
blob := compressBlob(data)
if err := os.WriteFile(path, blob, 0o644); err != nil {
return 0, fmt.Errorf("write blob: %w", err)
}
return int64(len(blob)), nil
}
// writeManifestPair writes a NSYM manifest and its .json sidecar. The caller
// supplies the sidecar fields it knows; the rest are derived from the entries.
// data must be the serialised form of entries.
func writeManifestPair(manifestPath string, data []byte, entries []Entry, onDiskBytes int64, sidecar Sidecar) error {
if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil {
return fmt.Errorf("create manifest directory: %w", err)
}
if err := os.WriteFile(manifestPath, data, 0o644); err != nil {
return fmt.Errorf("write manifest: %w", err)
}
var totalBytes int64 var totalBytes int64
clientContents := false clientContents := false
for _, entry := range entries { for _, entry := range entries {
@@ -240,8 +260,5 @@ func writeManifestPair(manifestPath string, data []byte, entries []Entry, onDisk
if err != nil { if err != nil {
return err return err
} }
if err := os.WriteFile(manifestPath+".json", body, 0o644); err != nil { return target.putIndex(key, data, body)
return fmt.Errorf("write sidecar: %w", err)
}
return nil
} }
+65 -39
View File
@@ -3,6 +3,7 @@ package nwsync
import ( import (
"bytes" "bytes"
"crypto/sha1" "crypto/sha1"
"crypto/sha256"
"encoding/binary" "encoding/binary"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
@@ -45,6 +46,30 @@ func writeHak(t *testing.T, path string, contents map[string][]byte) {
} }
} }
// artifactKey is the depot key a file would be published under: the sha256 of
// its bytes, hash-tree depth 2, keeping the extension.
func artifactKey(t *testing.T, path string) string {
t.Helper()
body, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
sum := sha256.Sum256(body)
digest := hex.EncodeToString(sum[:])
return "artifacts/haks/sha256/" + digest[0:2] + "/" + digest[2:4] + "/" + digest + filepath.Ext(path)
}
// emitLocal emits one artifact into a local tree, the conformance path.
func emitLocal(t *testing.T, path, out string) (EmitResult, error) {
t.Helper()
return Emit(EmitOptions{
ArtifactKey: artifactKey(t, path),
ArtifactPath: path,
As: filepath.Base(path),
OutDir: out,
})
}
func TestBlobFramingRoundTrips(t *testing.T) { func TestBlobFramingRoundTrips(t *testing.T) {
data := []byte("the quick brown fox jumps over the lazy dog, repeatedly and at length") data := []byte("the quick brown fox jumps over the lazy dog, repeatedly and at length")
blob := compressBlob(data) blob := compressBlob(data)
@@ -143,7 +168,7 @@ func TestEmitWritesBlobsAndManifest(t *testing.T) {
}) })
out := filepath.Join(dir, "out") out := filepath.Join(dir, "out")
result, err := Emit(hak, out) result, err := emitLocal(t, hak, out)
if err != nil { if err != nil {
t.Fatalf("emit: %v", err) t.Fatalf("emit: %v", err)
} }
@@ -171,7 +196,7 @@ func TestEmitWritesBlobsAndManifest(t *testing.T) {
t.Errorf("blob decompressed to %q, want %q", got, body) t.Errorf("blob decompressed to %q, want %q", got, body)
} }
entries := readEmitted(t, out, "sow_test_01") entries := readEmitted(t, result.ManifestPath)
for _, entry := range entries { for _, entry := range entries {
if entry.ResType == restype(t, "nss") || entry.ResType == restype(t, "ndb") || entry.ResType == restype(t, "gic") { if entry.ResType == restype(t, "nss") || entry.ResType == restype(t, "ndb") || entry.ResType == restype(t, "gic") {
t.Errorf("skipped restype leaked into the manifest: %+v", entry) t.Errorf("skipped restype leaked into the manifest: %+v", entry)
@@ -200,9 +225,9 @@ func TestEmitWritesBlobsAndManifest(t *testing.T) {
} }
} }
func readEmitted(t *testing.T, dir, name string) []Entry { func readEmitted(t *testing.T, indexPath string) []Entry {
t.Helper() t.Helper()
data, err := os.ReadFile(filepath.Join(dir, name+".nsym")) data, err := os.ReadFile(indexPath)
if err != nil { if err != nil {
t.Fatalf("read emitted manifest: %v", err) t.Fatalf("read emitted manifest: %v", err)
} }
@@ -218,7 +243,7 @@ func TestEmitFailsClosedOnOversizeResource(t *testing.T) {
hak := filepath.Join(dir, "big.hak") hak := filepath.Join(dir, "big.hak")
writeHak(t, hak, map[string][]byte{"huge1.tga": make([]byte, fileSizeLimit+1)}) writeHak(t, hak, map[string][]byte{"huge1.tga": make([]byte, fileSizeLimit+1)})
if _, err := Emit(hak, filepath.Join(dir, "out")); err == nil { if _, err := emitLocal(t, hak, filepath.Join(dir, "out")); err == nil {
t.Fatal("emit accepted a resource over the 15 MB limit") t.Fatal("emit accepted a resource over the 15 MB limit")
} }
} }
@@ -230,11 +255,11 @@ func TestEmitLooseFile(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
out := filepath.Join(dir, "out") out := filepath.Join(dir, "out")
result, err := Emit(tlk, out) result, err := emitLocal(t, tlk, out)
if err != nil { if err != nil {
t.Fatalf("emit tlk: %v", err) t.Fatalf("emit tlk: %v", err)
} }
entries := readEmitted(t, out, "sow_tlk") entries := readEmitted(t, result.ManifestPath)
if len(entries) != 1 || entries[0].ResRef != "sow_tlk" || entries[0].ResType != restype(t, "tlk") { if len(entries) != 1 || entries[0].ResRef != "sow_tlk" || entries[0].ResType != restype(t, "tlk") {
t.Fatalf("tlk emitted as %+v", entries) t.Fatalf("tlk emitted as %+v", entries)
} }
@@ -245,33 +270,34 @@ func TestEmitLooseFile(t *testing.T) {
// emitFixture emits two haks that share a resref, so the merge rule is // emitFixture emits two haks that share a resref, so the merge rule is
// observable: "top" holds the winning body, "assets" the shadowed one. // observable: "top" holds the winning body, "assets" the shadowed one.
func emitFixture(t *testing.T) (string, []byte, []byte) { func emitFixture(t *testing.T) (out string, keys map[string]string, topBody, assetBody []byte) {
t.Helper() t.Helper()
dir := t.TempDir() dir := t.TempDir()
topBody := []byte("2da from sow_top") topBody = []byte("2da from sow_top")
assetBody := []byte("2da from the asset hak") assetBody = []byte("2da from the asset hak")
writeHak(t, filepath.Join(dir, "sow_top.hak"), map[string][]byte{"appearance.2da": topBody}) writeHak(t, filepath.Join(dir, "sow_top.hak"), map[string][]byte{"appearance.2da": topBody})
writeHak(t, filepath.Join(dir, "sow_core_01.hak"), map[string][]byte{ writeHak(t, filepath.Join(dir, "sow_core_01.hak"), map[string][]byte{
"appearance.2da": assetBody, "appearance.2da": assetBody,
"bloodstain1.tga": []byte("blood"), "bloodstain1.tga": []byte("blood"),
}) })
out := filepath.Join(dir, "out") out = filepath.Join(dir, "out")
keys = map[string]string{}
for _, name := range []string{"sow_top", "sow_core_01"} { for _, name := range []string{"sow_top", "sow_core_01"} {
if _, err := Emit(filepath.Join(dir, name+".hak"), out); err != nil { path := filepath.Join(dir, name+".hak")
keys[name] = artifactKey(t, path)
if _, err := emitLocal(t, path, out); err != nil {
t.Fatalf("emit %s: %v", name, err) t.Fatalf("emit %s: %v", name, err)
} }
} }
return out, topBody, assetBody return out, keys, topBody, assetBody
} }
func TestAssembleShadowsByOrder(t *testing.T) { func TestAssembleShadowsByOrder(t *testing.T) {
entriesDir, topBody, assetBody := emitFixture(t) entriesDir, keys, topBody, assetBody := emitFixture(t)
out := t.TempDir()
result, err := Assemble(AssembleOptions{ result, err := Assemble(AssembleOptions{
Order: []string{"sow_top", "sow_core_01"}, ArtifactKeys: []string{keys["sow_top"], keys["sow_core_01"]},
EntriesDir: entriesDir, OutDir: entriesDir,
OutDir: out,
GroupID: 2, GroupID: 2,
}) })
if err != nil { if err != nil {
@@ -325,13 +351,11 @@ func TestAssembleShadowsByOrder(t *testing.T) {
} }
func TestAssembleReversedOrderPicksTheOtherHak(t *testing.T) { func TestAssembleReversedOrderPicksTheOtherHak(t *testing.T) {
entriesDir, topBody, assetBody := emitFixture(t) entriesDir, keys, topBody, assetBody := emitFixture(t)
out := t.TempDir()
result, err := Assemble(AssembleOptions{ result, err := Assemble(AssembleOptions{
Order: []string{"sow_core_01", "sow_top"}, ArtifactKeys: []string{keys["sow_core_01"], keys["sow_top"]},
EntriesDir: entriesDir, OutDir: entriesDir,
OutDir: out,
}) })
if err != nil { if err != nil {
t.Fatalf("assemble: %v", err) t.Fatalf("assemble: %v", err)
@@ -350,9 +374,13 @@ func TestAssembleReversedOrderPicksTheOtherHak(t *testing.T) {
} }
func TestAssembleRefusesMismatchedEmitterVersions(t *testing.T) { func TestAssembleRefusesMismatchedEmitterVersions(t *testing.T) {
entriesDir, _, _ := emitFixture(t) entriesDir, keys, _, _ := emitFixture(t)
path := filepath.Join(entriesDir, "sow_core_01.nsym.json") index, err := resolveIndexKey(keys["sow_core_01"], entriesDir)
if err != nil {
t.Fatal(err)
}
path := filepath.Join(entriesDir, index+".json")
var sidecar Sidecar var sidecar Sidecar
body, err := os.ReadFile(path) body, err := os.ReadFile(path)
if err != nil { if err != nil {
@@ -371,9 +399,8 @@ func TestAssembleRefusesMismatchedEmitterVersions(t *testing.T) {
} }
_, err = Assemble(AssembleOptions{ _, err = Assemble(AssembleOptions{
Order: []string{"sow_top", "sow_core_01"}, ArtifactKeys: []string{keys["sow_top"], keys["sow_core_01"]},
EntriesDir: entriesDir, OutDir: entriesDir,
OutDir: t.TempDir(),
}) })
if err == nil || !strings.Contains(err.Error(), "emitter version mismatch") { if err == nil || !strings.Contains(err.Error(), "emitter version mismatch") {
t.Fatalf("assemble merged across emitter versions: %v", err) t.Fatalf("assemble merged across emitter versions: %v", err)
@@ -385,7 +412,7 @@ func TestEmitHonoursSourceDateEpoch(t *testing.T) {
dir := t.TempDir() dir := t.TempDir()
hak := filepath.Join(dir, "pinned.hak") hak := filepath.Join(dir, "pinned.hak")
writeHak(t, hak, map[string][]byte{"one1.tga": []byte("body")}) writeHak(t, hak, map[string][]byte{"one1.tga": []byte("body")})
result, err := Emit(hak, filepath.Join(dir, "out")) result, err := emitLocal(t, hak, filepath.Join(dir, "out"))
if err != nil { if err != nil {
t.Fatalf("emit: %v", err) t.Fatalf("emit: %v", err)
} }
@@ -414,19 +441,19 @@ func TestEmitRejectsAModule(t *testing.T) {
if err := os.WriteFile(path, out.Bytes(), 0o644); err != nil { if err := os.WriteFile(path, out.Bytes(), 0o644); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if _, err := Emit(path, filepath.Join(dir, "out")); err == nil { if _, err := emitLocal(t, path, filepath.Join(dir, "out")); err == nil {
t.Fatal("emit accepted a .mod; a manifest never carries module contents") t.Fatal("emit accepted a .mod; a manifest never carries module contents")
} }
} }
func TestAssembleFailsClosedOnMissingIndex(t *testing.T) { func TestAssembleFailsClosedOnMissingIndex(t *testing.T) {
entriesDir, _, _ := emitFixture(t) entriesDir, keys, _, _ := emitFixture(t)
missing := "artifacts/haks/sha256/00/11/" + strings.Repeat("0", 64) + ".hak"
_, err := Assemble(AssembleOptions{ _, err := Assemble(AssembleOptions{
Order: []string{"sow_top", "sow_never_published"}, ArtifactKeys: []string{keys["sow_top"], missing},
EntriesDir: entriesDir, OutDir: entriesDir,
OutDir: t.TempDir(),
}) })
if err == nil || !strings.Contains(err.Error(), "sow_never_published") { if err == nil || !strings.Contains(err.Error(), missing) {
t.Fatalf("assemble did not fail closed and name the missing artifact: %v", err) t.Fatalf("assemble did not fail closed and name the missing artifact: %v", err)
} }
} }
@@ -436,10 +463,9 @@ func TestRunUsageErrors(t *testing.T) {
nil, nil,
{"nope"}, {"nope"},
{"emit"}, {"emit"},
{"emit", "artifact.hak"}, {"emit", "artifact-key.hak"},
{"assemble", "--entries", "x", "--out", "y"}, {"emit", "a", "b", "c"},
{"assemble", "--order", "a", "--out", "y"}, {"assemble", "--out", "y"},
{"assemble", "--order", "a", "--entries", "x"},
} }
for _, args := range cases { for _, args := range cases {
var out, errw bytes.Buffer var out, errw bytes.Buffer
+50 -39
View File
@@ -12,7 +12,6 @@ import (
"flag" "flag"
"fmt" "fmt"
"io" "io"
"strings"
) )
const ( const (
@@ -45,37 +44,63 @@ func Run(args []string, stdout, stderr io.Writer) int {
func printRunUsage(w io.Writer) { func printRunUsage(w io.Writer) {
fmt.Fprint(w, `usage: fmt.Fprint(w, `usage:
nwsync emit <artifact> --out DIR nwsync emit [--as NAME] [--out DIR] <artifact-key> <file>
nwsync assemble --order NAMES --entries DIR --out DIR [--group-id N] 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 emit explodes one .hak/.erf or one loose file (the TLK) into NWSync blobs plus
a NSYM manifest covering only that artifact. assemble merges those per-artifact a NSYM index covering only that artifact, and uploads both. assemble merges
manifests into one, reading no bulk data. 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 { func runEmit(args []string, stdout, stderr io.Writer) int {
fs := flag.NewFlagSet("emit", flag.ContinueOnError) fs := flag.NewFlagSet("emit", flag.ContinueOnError)
fs.SetOutput(stderr) fs.SetOutput(stderr)
out := fs.String("out", "", "output directory (blobs plus the per-artifact manifest)") as := fs.String("as", "", "published name of the artifact, when it differs from the key")
if err := fs.Parse(args); err != nil { out := fs.String("out", "", "write to a local repository tree instead of uploading")
positional, err := parseArgs(fs, args)
if err != nil {
return exitUsage return exitUsage
} }
if fs.NArg() != 1 { if len(positional) != 2 {
fmt.Fprintf(stderr, "nwsync emit: exactly one artifact is required\n") fmt.Fprintf(stderr, "nwsync emit: <artifact-key> and <file> are both required\n")
return exitUsage
}
if *out == "" {
fmt.Fprintf(stderr, "nwsync emit: --out is required\n")
return exitUsage return exitUsage
} }
result, err := Emit(fs.Arg(0), *out) result, err := Emit(EmitOptions{
ArtifactKey: positional[0],
ArtifactPath: positional[1],
As: *as,
OutDir: *out,
})
if err != nil { if err != nil {
fmt.Fprintf(stderr, "nwsync emit: %v\n", err) fmt.Fprintf(stderr, "nwsync emit: %v\n", err)
return exitInternal return exitInternal
} }
fmt.Fprintf(stdout, "emitted %s: %d resources, %d new blobs, manifest %s\n", fmt.Fprintf(stdout, "emitted %s: %d resources, %d new blobs, index %s\n",
result.Name, result.Entries, result.BlobsWritten, result.ManifestPath) result.Name, result.Entries, result.BlobsWritten, result.ManifestPath)
return exitOK return exitOK
} }
@@ -83,37 +108,23 @@ func runEmit(args []string, stdout, stderr io.Writer) int {
func runAssemble(args []string, stdout, stderr io.Writer) int { func runAssemble(args []string, stdout, stderr io.Writer) int {
fs := flag.NewFlagSet("assemble", flag.ContinueOnError) fs := flag.NewFlagSet("assemble", flag.ContinueOnError)
fs.SetOutput(stderr) fs.SetOutput(stderr)
order := fs.String("order", "", "comma-separated artifact names, highest priority first") tlkKey := fs.String("tlk-key", "", "depot key of the TLK, which shadows nothing and merges last")
entries := fs.String("entries", "", "directory holding the per-artifact .nsym files") out := fs.String("out", "", "write to a local repository tree instead of uploading")
out := fs.String("out", "", "repository root; the manifest lands in <out>/manifests")
groupID := fs.Int("group-id", 0, "NWSync group id (1 = current, 2 = testing; 0 omits it)") 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") moduleName := fs.String("module-name", "", "module name recorded in the sidecar")
description := fs.String("description", "", "description recorded in the sidecar") description := fs.String("description", "", "description recorded in the sidecar")
if err := fs.Parse(args); err != nil { positional, err := parseArgs(fs, args)
if err != nil {
return exitUsage return exitUsage
} }
switch { if len(positional) == 0 {
case *order == "": fmt.Fprintf(stderr, "nwsync assemble: at least one artifact key is required\n")
fmt.Fprintf(stderr, "nwsync assemble: --order is required\n")
return exitUsage return exitUsage
case *entries == "":
fmt.Fprintf(stderr, "nwsync assemble: --entries is required\n")
return exitUsage
case *out == "":
fmt.Fprintf(stderr, "nwsync assemble: --out is required\n")
return exitUsage
}
names := make([]string, 0, 8)
for _, name := range strings.Split(*order, ",") {
if name = strings.TrimSpace(name); name != "" {
names = append(names, name)
}
} }
result, err := Assemble(AssembleOptions{ result, err := Assemble(AssembleOptions{
Order: names, ArtifactKeys: positional,
EntriesDir: *entries, TLKKey: *tlkKey,
OutDir: *out, OutDir: *out,
GroupID: *groupID, GroupID: *groupID,
ModuleName: *moduleName, ModuleName: *moduleName,
@@ -123,7 +134,7 @@ func runAssemble(args []string, stdout, stderr io.Writer) int {
fmt.Fprintf(stderr, "nwsync assemble: %v\n", err) fmt.Fprintf(stderr, "nwsync assemble: %v\n", err)
return exitInternal return exitInternal
} }
fmt.Fprintf(stdout, "assembled %s: %d resources from %d artifacts\n", fmt.Fprintf(stdout, "assembled manifest %s: %d resources, %s\n",
result.SHA1, result.Entries, len(names)) result.SHA1, result.Entries, result.ManifestPath)
return exitOK return exitOK
} }
+207
View File
@@ -0,0 +1,207 @@
package nwsync
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"os"
"path"
"path/filepath"
"strings"
"git.westgate.pw/ShadowsOverWestgate/sow-tools/internal/depot"
)
// sink is where an emit or assemble run puts what it produces. The zone is the
// production sink; a local directory exists only as the conformance path, so
// upstream's output and ours can be diffed on a developer machine.
type sink interface {
// putBlob stores one NWCompressedBuffer blob under its sha1 name and
// returns the bytes stored, or 0 if the blob was already there. Blob names
// are content hashes, so an existing name is existing content — which is
// why body is a thunk: compression is the expensive part of emit and a
// blob that is already stored must not pay for it.
putBlob(sha1Hex string, body func() []byte) (int64, error)
// putIndex stores a NSYM manifest and its sidecar under key, which is
// either an artifact-derived object key or a local path.
putIndex(key string, manifest, sidecar []byte) error
// getIndex reads back a NSYM manifest and its sidecar.
getIndex(key string) (manifest, sidecar []byte, err error)
// describe names the sink for messages.
describe(key string) string
}
// dirSink writes a local NWSync repository tree.
type dirSink struct{ root string }
func (s dirSink) putBlob(sha1Hex string, body func() []byte) (int64, error) {
blob := blobPath(s.root, sha1Hex)
if _, err := os.Stat(blob); err == nil {
return 0, nil
}
if err := os.MkdirAll(filepath.Dir(blob), 0o755); err != nil {
return 0, fmt.Errorf("create blob directory: %w", err)
}
data := body()
if err := os.WriteFile(blob, data, 0o644); err != nil {
return 0, fmt.Errorf("write blob: %w", err)
}
return int64(len(data)), nil
}
func (s dirSink) putIndex(key string, manifest, sidecar []byte) error {
target := filepath.Join(s.root, filepath.FromSlash(key))
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return fmt.Errorf("create manifest directory: %w", err)
}
if err := os.WriteFile(target, manifest, 0o644); err != nil {
return fmt.Errorf("write manifest: %w", err)
}
if err := os.WriteFile(target+".json", sidecar, 0o644); err != nil {
return fmt.Errorf("write sidecar: %w", err)
}
return nil
}
func (s dirSink) getIndex(key string) ([]byte, []byte, error) {
target := filepath.Join(s.root, filepath.FromSlash(key))
manifest, err := os.ReadFile(target)
if err != nil {
return nil, nil, fmt.Errorf("read index: %w", err)
}
sidecar, err := os.ReadFile(target + ".json")
if err != nil {
return nil, nil, fmt.Errorf("read sidecar: %w", err)
}
return manifest, sidecar, nil
}
func (s dirSink) describe(key string) string {
return filepath.Join(s.root, filepath.FromSlash(key))
}
// zoneSink uploads straight to the NWSync storage zone. Nothing bulky is ever
// written to the runner's disk: the working set is one resource at a time.
type zoneSink struct {
store depot.KeyStore
ctx context.Context
zone string
}
func (s zoneSink) putBlob(sha1Hex string, body func() []byte) (int64, error) {
key := path.Join("data", "sha1", sha1Hex[0:2], sha1Hex[2:4], sha1Hex)
// A throttled probe must never be read as "missing, re-upload" or as
// "present, skip", so only a confirmed Present skips the upload.
state, _, err := s.store.ProbeKey(s.ctx, key)
if err != nil {
return 0, fmt.Errorf("probe blob %s: %w", sha1Hex, err)
}
if state == depot.Present {
return 0, nil
}
data := body()
if err := s.put(key, data); err != nil {
return 0, fmt.Errorf("upload blob %s: %w", sha1Hex, err)
}
return int64(len(data)), nil
}
func (s zoneSink) putIndex(key string, manifest, sidecar []byte) error {
// The manifest lands last: its presence is the publication marker, so it
// must never appear before the blobs it names.
if err := s.put(key+".json", sidecar); err != nil {
return fmt.Errorf("upload sidecar %s: %w", key, err)
}
if err := s.put(key, manifest); err != nil {
return fmt.Errorf("upload index %s: %w", key, err)
}
return nil
}
func (s zoneSink) getIndex(key string) ([]byte, []byte, error) {
manifest, err := s.store.GetKey(s.ctx, key)
if err != nil {
return nil, nil, fmt.Errorf("read index: %w", err)
}
sidecar, err := s.store.GetKey(s.ctx, key+".json")
if err != nil {
return nil, nil, fmt.Errorf("read sidecar: %w", err)
}
return manifest, sidecar, nil
}
func (s zoneSink) describe(key string) string { return s.zone + "/" + key }
func (s zoneSink) put(key string, body []byte) error {
sum := sha256.Sum256(body)
return s.store.PutReader(s.ctx, key, bytes.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:]))
}
// newZoneSink builds the upload sink from the environment. NWSync data lives
// in its own zone, separate from the asset depot, so it has its own zone and
// credential; only the host is shared, and Crucible has no default host.
func newZoneSink(ctx context.Context, getenv func(string) string) (sink, error) {
cfg := depot.LoadConfig(getenv)
cfg.StorageZone = getenv("NWSYNC_STORAGE_ZONE")
cfg.WriteKey = getenv("NWSYNC_STORAGE_PASSWORD")
cfg.ReadKey = cfg.WriteKey
if cfg.StorageZone == "" {
return nil, fmt.Errorf("NWSYNC_STORAGE_ZONE is unset (or pass --out DIR to write locally)")
}
if cfg.WriteKey == "" {
return nil, fmt.Errorf("NWSYNC_STORAGE_PASSWORD is unset (or pass --out DIR to write locally)")
}
if cfg.StorageHost == "" {
return nil, fmt.Errorf("BUNNY_STORAGE_HOST is unset")
}
store, err := depot.NewKeyStore(cfg)
if err != nil {
return nil, err
}
return zoneSink{store: store, ctx: ctx, zone: cfg.StorageZone}, nil
}
// indexKey is where an artifact's NSYM lives: beside the artifact itself, with
// the final extension replaced. emit and assemble must agree on this one rule,
// so it lives here and nowhere else.
//
// artifacts/haks/sha256/30/46/3046….hak -> artifacts/haks/sha256/30/46/3046….nsym
func indexKey(artifactKey string) (string, error) {
extension := path.Ext(artifactKey)
if extension == "" {
return "", fmt.Errorf("artifact key %q has no extension", artifactKey)
}
return strings.TrimSuffix(artifactKey, extension) + ".nsym", nil
}
// resolveIndexKey is where emit writes an artifact's index and where assemble
// reads it from. On the zone that is beside the artifact; locally the indexes
// sit flat beside the data tree, so upstream's output and ours diff directly.
func resolveIndexKey(artifactKey, outDir string) (string, error) {
key, err := indexKey(artifactKey)
if err != nil {
return "", err
}
if outDir != "" {
return path.Base(key), nil
}
return key, nil
}
// checkArtifactKey fails closed when the key's embedded digest is not the
// digest of the bytes being emitted. Publishing an index under the wrong key
// silently pairs a manifest with the wrong artifact.
func checkArtifactKey(artifactKey string, artifact []byte) error {
base := path.Base(artifactKey)
digest := strings.TrimSuffix(base, path.Ext(base))
if len(digest) != 64 {
return fmt.Errorf("artifact key %q does not name a sha256", artifactKey)
}
sum := sha256.Sum256(artifact)
if got := hex.EncodeToString(sum[:]); got != digest {
return fmt.Errorf("artifact key %q names digest %s but the file hashes to %s", artifactKey, digest, got)
}
return nil
}
+249
View File
@@ -0,0 +1,249 @@
package nwsync
import (
"crypto/sha1"
"crypto/sha256"
"encoding/hex"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
)
// fakeZone is a Bunny-shaped object store: PUT stores, GET reads, and the
// Checksum header is verified the way Bunny verifies it.
type fakeZone struct {
mu sync.Mutex
objects map[string][]byte
puts []string
failOn func(key string) bool // when true, the PUT fails
}
func newFakeZone(t *testing.T) (*fakeZone, func(string) string) {
t.Helper()
zone := &fakeZone{objects: map[string][]byte{}}
server := httptest.NewServer(zone)
t.Cleanup(server.Close)
getenv := func(name string) string {
switch name {
case "NWSYNC_STORAGE_ZONE":
return "sow-nwsync"
case "NWSYNC_STORAGE_PASSWORD":
return "write-key"
case "BUNNY_STORAGE_HOST":
return server.URL
}
return ""
}
return zone, getenv
}
func (z *fakeZone) ServeHTTP(w http.ResponseWriter, r *http.Request) {
key := strings.TrimPrefix(r.URL.Path, "/sow-nwsync/")
switch r.Method {
case http.MethodPut:
if z.failOn != nil && z.failOn(key) {
http.Error(w, "boom", http.StatusInternalServerError)
return
}
body, err := io.ReadAll(r.Body)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
sum := sha256.Sum256(body)
if want := strings.ToUpper(hex.EncodeToString(sum[:])); r.Header.Get("Checksum") != want {
http.Error(w, "checksum mismatch", http.StatusBadRequest)
return
}
z.mu.Lock()
z.objects[key] = body
z.puts = append(z.puts, key)
z.mu.Unlock()
w.WriteHeader(http.StatusCreated)
case http.MethodGet:
z.mu.Lock()
body, ok := z.objects[key]
z.mu.Unlock()
if !ok {
http.Error(w, "not found", http.StatusNotFound)
return
}
_, _ = w.Write(body)
default:
http.Error(w, "unsupported", http.StatusMethodNotAllowed)
}
}
func (z *zoneSinkFixture) emit(t *testing.T, path string) EmitResult {
t.Helper()
result, err := Emit(EmitOptions{
ArtifactKey: artifactKey(t, path),
ArtifactPath: path,
Sink: z.sink,
})
if err != nil {
t.Fatalf("emit %s: %v", path, err)
}
return result
}
type zoneSinkFixture struct {
zone *fakeZone
sink sink
}
func newZoneFixture(t *testing.T) *zoneSinkFixture {
t.Helper()
zone, getenv := newFakeZone(t)
target, err := newZoneSink(t.Context(), getenv)
if err != nil {
t.Fatalf("zone sink: %v", err)
}
return &zoneSinkFixture{zone: zone, sink: target}
}
func sha1Of(body []byte) [20]byte { return sha1.Sum(body) }
func TestEmitUploadsBlobsThenIndex(t *testing.T) {
fixture := newZoneFixture(t)
dir := t.TempDir()
hak := filepath.Join(dir, "sow_test_01.hak")
body := []byte("texture bytes")
writeHak(t, hak, map[string][]byte{"bloodstain1.tga": body, "copy1.txi": body})
key := artifactKey(t, hak)
result := fixture.emit(t, hak)
index, err := indexKey(key)
if err != nil {
t.Fatal(err)
}
if _, ok := fixture.zone.objects[index]; !ok {
t.Fatalf("no index at %s; zone holds %v", index, fixture.zone.puts)
}
if result.BlobsWritten != 1 {
t.Errorf("uploaded %d blobs, want 1 (identical content shares a blob)", result.BlobsWritten)
}
// The index is the publication marker, so it must land after every blob it
// names — including its own sidecar.
last := fixture.zone.puts[len(fixture.zone.puts)-1]
if last != index {
t.Errorf("index landed at position %d of %d; it must be last", len(fixture.zone.puts), len(fixture.zone.puts))
}
for _, key := range fixture.zone.puts[:len(fixture.zone.puts)-1] {
if strings.HasPrefix(key, "data/sha1/") || key == index+".json" {
continue
}
t.Errorf("unexpected object uploaded before the index: %s", key)
}
}
func TestEmitSkipsBlobsAlreadyInTheZone(t *testing.T) {
fixture := newZoneFixture(t)
dir := t.TempDir()
hak := filepath.Join(dir, "sow_test_01.hak")
writeHak(t, hak, map[string][]byte{"bloodstain1.tga": []byte("blood")})
first := fixture.emit(t, hak)
if first.BlobsWritten != 1 {
t.Fatalf("first emit uploaded %d blobs, want 1", first.BlobsWritten)
}
second := fixture.emit(t, hak)
if second.BlobsWritten != 0 {
t.Errorf("re-emit uploaded %d blobs, want 0 (a blob name is its content)", second.BlobsWritten)
}
}
func TestEmitLeavesNoIndexWhenAnUploadFails(t *testing.T) {
fixture := newZoneFixture(t)
fixture.zone.failOn = func(key string) bool { return strings.HasPrefix(key, "data/sha1/") }
dir := t.TempDir()
hak := filepath.Join(dir, "sow_test_01.hak")
writeHak(t, hak, map[string][]byte{"bloodstain1.tga": []byte("blood")})
_, err := Emit(EmitOptions{
ArtifactKey: artifactKey(t, hak),
ArtifactPath: hak,
Sink: fixture.sink,
})
if err == nil {
t.Fatal("emit reported success after an upload failed")
}
for key := range fixture.zone.objects {
if strings.HasSuffix(key, ".nsym") {
t.Errorf("a half-emitted artifact published an index: %s", key)
}
}
}
func TestEmitRejectsAKeyThatDoesNotMatchTheFile(t *testing.T) {
fixture := newZoneFixture(t)
dir := t.TempDir()
hak := filepath.Join(dir, "sow_test_01.hak")
writeHak(t, hak, map[string][]byte{"bloodstain1.tga": []byte("blood")})
wrong := "artifacts/haks/sha256/00/11/" + strings.Repeat("0", 64) + ".hak"
_, err := Emit(EmitOptions{ArtifactKey: wrong, ArtifactPath: hak, Sink: fixture.sink})
if err == nil || !strings.Contains(err.Error(), "hashes to") {
t.Fatalf("emit published under a key that names another artifact: %v", err)
}
}
func TestAssembleReadsIndexesFromTheZone(t *testing.T) {
fixture := newZoneFixture(t)
dir := t.TempDir()
topBody := []byte("2da from sow_top")
assetBody := []byte("2da from the asset hak")
top := filepath.Join(dir, "sow_top.hak")
core := filepath.Join(dir, "sow_core_01.hak")
writeHak(t, top, map[string][]byte{"appearance.2da": topBody})
writeHak(t, core, map[string][]byte{"appearance.2da": assetBody, "bloodstain1.tga": []byte("blood")})
tlkPath := filepath.Join(dir, "sow_tlk.tlk")
if err := os.WriteFile(tlkPath, []byte("TLK V3.0 payload"), 0o644); err != nil {
t.Fatal(err)
}
fixture.emit(t, top)
fixture.emit(t, core)
if _, err := Emit(EmitOptions{
ArtifactKey: artifactKey(t, tlkPath),
ArtifactPath: tlkPath,
As: "sow_tlk.tlk",
Sink: fixture.sink,
}); err != nil {
t.Fatalf("emit tlk: %v", err)
}
result, err := Assemble(AssembleOptions{
ArtifactKeys: []string{artifactKey(t, top), artifactKey(t, core)},
TLKKey: artifactKey(t, tlkPath),
GroupID: 2,
Sink: fixture.sink,
})
if err != nil {
t.Fatalf("assemble: %v", err)
}
if result.Entries != 3 {
t.Fatalf("merged %d entries, want 3 (appearance.2da is shadowed, the TLK adds one)", result.Entries)
}
manifest, ok := fixture.zone.objects["manifests/"+result.SHA1]
if !ok {
t.Fatalf("no merged manifest in the zone; it holds %v", fixture.zone.puts)
}
entries, err := readManifest(manifest)
if err != nil {
t.Fatalf("parse merged manifest: %v", err)
}
for _, entry := range entries {
if entry.ResRef == "appearance" && entry.SHA1 != sha1Of(topBody) {
t.Errorf("appearance.2da resolved to the shadowed hak, not the first one given")
}
}
}