Date: Sun, 15 Nov 2026 01:38:41 +0100
Subject: [PATCH 5/5] fix(peernet): name bundle files by virtual path and bound
part shapes
Found by a live two-instance run and a review of the layout changes.
- bundle files are named after their virtual path; archive content had
no name, so every archive release was announced and then rejected
- the publisher runs the receiver's file policy before announcing
- real-pipeline test: importer, builder, verify, accept and read-back for
plain, generic-container, RAR4/RAR5 single and multi-volume and
password-in-NZB releases
- parts after the first must start near their article's beginning and
parts before the last must end near its end, which rejects files
stitched from small windows
- the queue row id is resolved after an upsert, fixing history writes
for direct peer imports
- local-policy rejections refund the per-publisher fetch budget, and the
default budget suits browsing
- the importer is not installed when a required collaborator is missing
- catch-up publishes are counted; status reports header-bound parts
- current rules consolidated in the spec
---
cmd/altmount/cmd/peernet_bundle_wiring.go | 38 ++-
.../cmd/peernet_bundle_wiring_test.go | 18 +-
docs/docs/3. Configuration/peer-network.md | 7 +-
.../specs/2026-11-12-peer-network-design.md | 45 +++-
.../config/peernet/ActivityCard.tsx | 2 +
.../components/config/peernet/peernetStats.ts | 21 +-
frontend/src/types/peernet.ts | 5 +
frontend/tests/peernetStats.test.ts | 25 ++
internal/api/peernet_handlers.go | 9 +-
internal/database/queue_addtoqueue_id_test.go | 56 ++++
internal/database/queue_repository.go | 15 +-
.../importer/peer_bundle_realpipeline_test.go | 244 ++++++++++++++++++
internal/importer/peer_bundle_service_test.go | 23 ++
.../archivehdr/archivetest/archivetest.go | 26 +-
.../archivehdr/archivetest/rar4volume_test.go | 48 ++++
internal/peernet/bridge/bundleimport.go | 9 +
internal/peernet/bundle/limits.go | 14 +
internal/peernet/bundle/poc_regress_test.go | 169 ++++++++++++
internal/peernet/bundle/validate.go | 35 ++-
internal/peernet/bundleimport/accept.go | 18 +-
internal/peernet/bundleimport/accept_test.go | 6 +-
internal/peernet/bundleimport/filepolicy.go | 8 +-
internal/peernet/bundleimport/importer.go | 34 ++-
.../peernet/bundleimport/newchecked_test.go | 24 ++
internal/peernet/bundleimport/parts.go | 6 +-
internal/peernet/bundleimport/pathobserver.go | 2 +-
.../peernet/bundleimport/pathobserver_test.go | 30 +++
internal/peernet/bundleimport/refund_test.go | 43 +++
internal/peernet/bundleimport/types.go | 45 +++-
internal/peernet/bundlex/build_bench_test.go | 4 +-
internal/peernet/bundlex/builder.go | 10 +-
internal/peernet/bundlex/builder_test.go | 34 ++-
internal/peernet/bundlex/fetch.go | 8 +
internal/peernet/bundlex/fetch_test.go | 28 +-
internal/peernet/bundlex/limits.go | 2 +-
internal/peernet/service/bundles.go | 24 ++
internal/peernet/service/catchup_test.go | 3 +
internal/peernet/service/publishing.go | 2 +
internal/testsupport/mediagen/mediagen.go | 56 ++++
39 files changed, 1132 insertions(+), 64 deletions(-)
create mode 100644 internal/database/queue_addtoqueue_id_test.go
create mode 100644 internal/importer/peer_bundle_realpipeline_test.go
create mode 100644 internal/peernet/archivehdr/archivetest/rar4volume_test.go
create mode 100644 internal/peernet/bundle/poc_regress_test.go
create mode 100644 internal/peernet/bundleimport/newchecked_test.go
create mode 100644 internal/peernet/bundleimport/refund_test.go
create mode 100644 internal/testsupport/mediagen/mediagen.go
diff --git a/cmd/altmount/cmd/peernet_bundle_wiring.go b/cmd/altmount/cmd/peernet_bundle_wiring.go
index 6d1c5be7a..623469074 100644
--- a/cmd/altmount/cmd/peernet_bundle_wiring.go
+++ b/cmd/altmount/cmd/peernet_bundle_wiring.go
@@ -163,16 +163,24 @@ type peernetBundleHost struct {
hasPending func() bool
// observe installs the importer's path observer on the metadata service.
observe func(metadata.FileObserver)
- build func() *bundleimport.Importer
+ build func() (*bundleimport.Importer, error)
getCfg config.ConfigGetter
mu sync.Mutex
imp *bundleimport.Importer
}
+// importer returns the bundle importer, building it on first use. It returns
+// nil (and logs why) when a required collaborator is missing: such an importer
+// is never installed, so callers must treat nil as "feature unavailable".
func (h *peernetBundleHost) importer() *bundleimport.Importer {
if h.imp == nil {
- h.imp = h.build()
+ imp, err := h.build()
+ if err != nil {
+ slog.Error("Peer bundle import is not available", "error", err)
+ return nil
+ }
+ h.imp = imp
if h.observe != nil {
h.observe(h.imp.PathObserver())
}
@@ -205,15 +213,21 @@ func (h *peernetBundleHost) apply(cfg *config.Config) {
if !cfg.PeernetEnabled() {
h.importerSvc.SetPeerBundleFastPath(nil)
if h.hasPending != nil && h.hasPending() {
- h.health.SetPeerBundleGate(h.importer())
- h.api.SetPeerBundleLookup(h.lookup)
- return
+ if imp := h.importer(); imp != nil {
+ h.health.SetPeerBundleGate(imp)
+ h.api.SetPeerBundleLookup(h.lookup)
+ return
+ }
}
h.health.SetPeerBundleGate(nil)
h.api.SetPeerBundleLookup(nil)
return
}
imp := h.importer()
+ if imp == nil {
+ h.importerSvc.SetPeerBundleFastPath(nil)
+ return
+ }
// Pending peer imports must be honoured even before a fetcher exists.
h.health.SetPeerBundleGate(imp)
h.api.SetPeerBundleLookup(h.lookup)
@@ -290,14 +304,8 @@ func newPeernetBundleHost(d bundleWiringDeps, seams *peernetBundleSeams) *peerne
n, _, err := state.PendingCounts(context.Background(), time.Time{})
return err != nil || n > 0
},
- build: func() *bundleimport.Importer {
- cfg := bundleImporterConfig(d, state, stagingRoot, seams)
- if missing := cfg.Missing(); len(missing) > 0 {
- // A programming error: the accept pipeline would reject every
- // bundle at runtime. Say so loudly instead of failing quietly.
- slog.Error("Peer bundle import is wired without required collaborators", "missing", missing)
- }
- return bundleimport.New(cfg)
+ build: func() (*bundleimport.Importer, error) {
+ return bundleimport.NewChecked(bundleImporterConfig(d, state, stagingRoot, seams))
},
}
}
@@ -381,5 +389,7 @@ func (b *peerOriginBinder) ImportStatus() service.BundleImportStatus {
pending, stale := h.Importer().PendingStatus(context.Background())
return service.BundleImportStatus{Pending: int64(pending), Stale: int64(stale), Accepted: int64(c.Accepted), AlreadyImported: int64(c.AlreadyImported), Rejected: rej,
FullCheckPassed: int64(c.FullCheckPassed), FullCheckFailed: int64(c.FullCheckFailed),
- Announced: int64(c.Announced), BadBundleReports: int64(c.BadBundleReports)}
+ Announced: int64(c.Announced), BadBundleReports: int64(c.BadBundleReports),
+ PartsHeaderBound: int64(c.PartsHeaderBound), PartsHeaderUnbound: int64(c.PartsHeaderUnbound),
+ ExtentsHeaderBound: int64(c.ExtentsHeaderBound), ExtentsHeaderUnbound: int64(c.ExtentsHeaderUnbound)}
}
diff --git a/cmd/altmount/cmd/peernet_bundle_wiring_test.go b/cmd/altmount/cmd/peernet_bundle_wiring_test.go
index 5f26d18b9..ae235b574 100644
--- a/cmd/altmount/cmd/peernet_bundle_wiring_test.go
+++ b/cmd/altmount/cmd/peernet_bundle_wiring_test.go
@@ -41,7 +41,7 @@ func bundleHostFixture(t *testing.T) (*peernetBundleHost, *recFastPath, *recGate
fp, g, l, built := &recFastPath{}, &recGate{}, &recLookup{}, new(int)
h := &peernetBundleHost{
importerSvc: fp, health: g, api: l, seams: &peernetBundleSeams{}, receiving: func() bool { return true },
- build: func() *bundleimport.Importer { *built++; return bundleimport.New(bundleimport.Config{}) },
+ build: func() (*bundleimport.Importer, error) { *built++; return bundleimport.New(bundleimport.Config{}), nil },
}
return h, fp, g, l, built
}
@@ -136,7 +136,9 @@ func TestPeerOriginBinder_FailsClosedUntilTheImporterExists(t *testing.T) {
require.Error(t, err, "unbound")
h, _, _, _, _ := bundleHostFixture(t)
- h.build = func() *bundleimport.Importer { return bundleimport.New(bundleimport.Config{State: stubBundleState{}}) }
+ h.build = func() (*bundleimport.Importer, error) {
+ return bundleimport.New(bundleimport.Config{State: stubBundleState{}}), nil
+ }
b.Bind(h)
_, _, err = b.PeerBundleStatus(context.Background(), "peer.mkv")
require.Error(t, err, "importer not built yet (feature not applied)")
@@ -294,3 +296,15 @@ var (
_ health.PeerBundleGate = (*bundleimport.Importer)(nil)
_ health.PeerHeaderVerifier = (*bundleimport.Importer)(nil)
)
+
+// An importer that cannot be built (a required collaborator is missing) is
+// never installed, anywhere.
+func TestBundleHost_NeverInstallsAnImporterThatCannotBeBuilt(t *testing.T) {
+ h, fp, g, l, _ := bundleHostFixture(t)
+ h.build = func() (*bundleimport.Importer, error) { return bundleimport.NewChecked(bundleimport.Config{}) }
+ assert.NotPanics(t, func() { h.apply(enabledCfg(true)) })
+ assert.Nil(t, fp.cur)
+ assert.Nil(t, g.cur, "no gate: a nil importer behind the interface would panic at first use")
+ assert.Nil(t, l.cur)
+ assert.Nil(t, h.Importer())
+}
diff --git a/docs/docs/3. Configuration/peer-network.md b/docs/docs/3. Configuration/peer-network.md
index 733ee78c3..fca203568 100644
--- a/docs/docs/3. Configuration/peer-network.md
+++ b/docs/docs/3. Configuration/peer-network.md
@@ -166,7 +166,7 @@ A **full check** of every segment is then scheduled at high priority. Until it p
**What Sonarr and Radarr see while it is pending.** An item they grabbed through the SABnzbd-compatible API stays in the *queue* as work in progress (status `Downloading`, 99 percent) and is not in the *history*, so they never import content that is not verified yet. When the full check passes, the item moves to the history as completed and the ARR imports it as usual. When it fails, or when the 7-day limit below is reached, it appears in the history as **failed** with a short message ("Peer-supplied release failed its full health check and was removed"), so the ARR blocklists the release and searches again. An item you hid from the ARRs yourself (skip ARR notification) stays hidden in every state. The Queue page of AltMount shows the item as completed from the start, because the files are playable at once.
-**When a full check fails.** In every case the same single path runs and no ARR repair is ever attempted for the file: the queue item turns into a failed one, and every file of that import is removed (metadata, content index, library copy, health record and peer state), so a corrupt file does not stay in your library and the rest of the release cannot complete the import later. This holds for an item an ARR grabbed, for a release started from Stremio (which no ARR knows) and for an import hidden from the ARRs. Only files of a peer import that is in the pending state are ever removed; nothing else is touched. The publisher is **not trusted for bundle fetches for a while**: 30 minutes, doubling per repeated failure up to 6 hours, forgiven after a day or as soon as a later bundle from the same publisher passes its full check. This never affects record sync. It is mild on purpose: a failure can also mean your own providers lack articles.
+**When a full check fails.** A check that finds missing articles is not a verdict yet: like any health check it is retried on the health schedule (every 15 minutes by default, up to the configured maximum number of retries), because articles can still be propagating. Only when the last retry fails is the import failed. A contradiction between the layout and the articles' own headers is different: it fails the import at once. When it fails, in every case the same single path runs and no ARR repair is ever attempted for the file: the queue item turns into a failed one, and every file of that import is removed (metadata, content index, library copy, health record and peer state), so a corrupt file does not stay in your library and the rest of the release cannot complete the import later. This holds for an item an ARR grabbed, for a release started from Stremio (which no ARR knows) and for an import hidden from the ARRs. Only files of a peer import that is in the pending state are ever removed; nothing else is touched. The publisher is **not trusted for bundle fetches for a while**: 30 minutes, doubling per repeated failure up to 6 hours, forgiven after a day or as soon as a later bundle from the same publisher passes its full check. This never affects record sync. It is mild on purpose: a failure can also mean your own providers lack articles.
**Stale imports.** A pending import is held for at most 7 days: after that it is failed exactly as above (the publisher is not blamed, since no check ran), and the status page shows how many are pending and how many waited too long. Turning the peer network off does not strand pending imports: the full check keeps running for them until they finish.
@@ -189,6 +189,11 @@ Before anything is written the layout is also checked against the articles thems
- **Archive extent checks.** For every file read through outer archive volumes, each extent (outer volume, offset, length) is compared with the archive's own file header, read from your providers through the same hidden reader that will serve the file (so an encrypted outer volume is read decrypted with the bundle's key). The header that precedes the extent's data must be a stored (not compressed) file entry, not a service block or a directory and not itself encrypted; its data must start exactly at the extent's offset and be as long as the extent in that volume; the entry's unpacked size must equal the file's size; and its continuation flags must fit the extent's place (first part, middle part, last part). RAR 4, RAR 5 and 7z (plain end header) are understood. When the outer volume is itself a stored file of a wrapping archive and the volume begins in that archive's first article, the wrapping header is checked as well (stored, same size, and for RAR 5 the IV in the header must equal the bundle's). Up to 64 extents are checked (the first and last of the main media file always, the rest chosen at random). A contradiction is attributable to the publisher: the bundle is refused (`extent_mismatch`), the publisher is reported (fetch back-off only) and the layout goes on the negative cache for 7 days.
- **Multi-part files (archive volumes).** Each slice of a file stored across volumes is bound the same way: the volume is read from its first article, and a stored, unencrypted (or, for a keyed file, encrypted) file entry's data must start at the byte where the slice starts, be as long as the slice, belong to a file of the declared size, and carry continuation flags that fit the slice's place (first, middle, last). For RAR 5 the volume numbers must also run in the order of the slices, so two middle volumes cannot be swapped. A slice that claims to be a whole posted file of media data must not begin like an archive volume. A second file stored after the first in a volume is bound too when your NZB is at hand (the articles before it are taken to be as long as the first one, which its own offset has to confirm). Up to 64 slices are checked (the first and last of every media file always, the rest chosen at random). A contradiction is attributable: `extent_mismatch`, the publisher is reported, the layout goes on the negative cache. A slice that cannot be judged (a second file in a volume of an old-format import, a split 7-Zip set whose end header is in another volume, encoded 7z headers) is unverifiable and treated like an unverifiable extent below.
- **What stays unverifiable.** An extent whose archive cannot be judged is *unverifiable*: an unknown format (a disc image such as a Blu-ray ISO), a 7z end header that is compressed or encrypted, encrypted archive headers, a header walk that exceeds its budget, a damaged header chain, or a missing article. Such a file is accepted only when its media passes the structural check (Matroska, MP4 or transport stream) through the layout; a generic container in that situation is refused (`extent_unverifiable`), because readable windows prove nothing about misplaced bytes. Extents beyond the 64 checked are covered by sampling only. Files stored inline in a single archive volume (no outer volumes) are not extents and are verified by the head and structure checks only.
+- **Part shape.** A file stored across archive volumes is described as parts, one per volume. Apart from the first and the last part, a part must start within 64 KiB of its first article and stop within 64 KiB (or one eighth of its articles, for volumes that carry a recovery record) of the end of its last article, so a part can never be a sliver of a large article. A bundle that breaks this is refused whatever the header checks say.
+- **File names.** A shared file carries the name it has in your library (the last part of its path). Releases whose files would all be refused by a receiver's file policy (no media file) are not announced; they are counted as not shareable (`no_media_file`).
+- **Costs and limits.** Fetching a bundle may be tried 10 times in a burst and 60 times per hour per publisher (and 10 in a burst, 60 per hour for the whole node). A bundle refused by your file policy before any article was read does not use up the publisher's share, and a bundle on the failed-bundle list costs nothing.
+- **What is still checked by sampling only.** Order of the volumes is bound only where the format numbers its volumes (RAR 5); for RAR 4 and split 7z sets a swap of two middle volumes is not detected at accept, and at most 64 volumes are header-checked per accept. The forced full check reads further headers. These limits remain until the checks are made exhaustive.
+- **Status.** The activity card shows how many parts and nested extents of accepted bundles were tied to their archive header and how many were not.
- **Sampling.** The last article of every run of articles (volume or file) is always read: a size lie anywhere in a run shifts every later offset, so the last article's offset and total expose it. The main media file's runs also get their first article and random interior ones (one per 500 articles, at least three). The first and a random interior article of the other runs are read as far as the budget allows, at randomly chosen runs. The budget is one percent of the release size, at least 64 MiB and at most 1 GiB, shared with the archive header reads above and raised if the last articles alone need more; only a release whose last articles exceed 1 GiB (about 1,400 volumes) is refused (`too_many_tables`). Optional samples that run out of the time limit are dropped, mandatory ones are not. A single movie file reads about 5 articles (4 MB), a 50-volume release about 90 (67 MB), a 200-volume release about 280 (215 MB), a 500-volume release about 700 (540 MB), plus a few articles per checked archive volume. A movie stored in a multi-volume archive set reads the last article of every volume, the first article of every checked volume, one article per checked volume for its header and one interior article per 500 articles of the movie: about 165 articles (127 MB) for 50 volumes of 100 MB and about 390 articles (300 MB) for 200 volumes. Two lies that cancel out inside a run are found only when a sampled article lies between them. With your NZB at hand the first article's yEnc file name must also match the NZB entry it belongs to. The forced full check reads a further random sample of headers (1 percent of the articles, 20 to 500) and fails the import on a contradiction. It samples; it does not prove.
- **Media structure.** Besides the end of the file, the container is read at random positions inside the file (Matroska: three random 1 MiB windows must contain cluster data; MPEG-TS: eight random packet positions must be in sync; fragmented MP4 must end in a real `moof`+`mdat` pair or `mfra`). What remains unprovable: bytes inside MP4 `mdat`, inside a Matroska frame or between sampled positions.
- **Degraded means failed.** A peer release whose full check finds the media playable but with missing segments is failed and removed, never announced. When you supplied the NZB the queue item is not marked failed: it is imported normally from your NZB, once, so you get what a normal import would have produced. A release imported directly (Stremio peer play) is failed. A missing article alone does not count against the publisher (it can be propagation just after posting); only a contradiction between the layout and the articles' own headers does, and that layout is then ignored whoever offers it, for 7 days.
diff --git a/docs/superpowers/specs/2026-11-12-peer-network-design.md b/docs/superpowers/specs/2026-11-12-peer-network-design.md
index 3aa48ced2..d124b2459 100644
--- a/docs/superpowers/specs/2026-11-12-peer-network-design.md
+++ b/docs/superpowers/specs/2026-11-12-peer-network-design.md
@@ -462,6 +462,41 @@ above.
publisher; the fallback title drops a trailing media extension; the
consumers' node-ID waiter makes a final check under its lock.
+## Phase 3: current rules (authoritative)
+
+This section states what the code does today. Where an older section below says
+something different, that bullet is marked "(superseded - see Current rules)".
+
+- **Shareable:** every release the importer produced whose files pass the
+ receiver's pure file policy (a media file after sanitising, allowed
+ extensions). The publisher runs that policy at build time and counts a
+ release that fails it as not shareable (`no_media_file` or the policy's own
+ reason). A file is named by the basename of its virtual path, never by
+ `OriginalFilename` (archive-stored content has none).
+- **Archive sets:** stored RAR 4, RAR 5 (single and multi-volume) and encrypted
+ stored RAR 5 with the password in the NZB are shared as parts of one file.
+ Multi-volume stored sets are no longer "not shareable".
+- **Part shape (deterministic, `bundle.Validate`):** in a multi-part file only
+ the first part may start deep inside its first article and only the last may
+ end deep inside its last one. Every other part starts within `HeaderSlack`
+ (64 KiB) of its first article and stops within the larger of `TrailerSlack`
+ (64 KiB) and one eighth of its articles from the end of its last article. A
+ part cannot be a sliver of a large article.
+- **Open network:** archive layouts are accepted and header-checked like on
+ private networks; the earlier "plain layouts only" restriction is gone.
+- **Fetch budget:** per publisher a burst of 10 and 60 per hour; node-wide
+ burst 10 and 60 per hour. A bundle the local file policy refused before any
+ article was read gives its per-publisher attempt back; a failed-bundle cache
+ hit costs nothing.
+- **Status:** the import status reports how many parts and extents of accepted
+ bundles were bound to an archive header and how many were not.
+- **Full check:** missing articles are retried on the health schedule before
+ the import fails; a header contradiction fails it at once.
+- **Still sampling-based (not done in this round):** accept reads at most 64
+ header checks, part order is bound only for RAR 5 volume numbers, verification
+ is charged by declared sizes, the fast-path window is fixed, and the forced
+ full check re-derives part boundaries from flat metadata.
+
## Phase 3: bundle format
Implemented by `internal/peernet/bundle`: a pure package (no network, database
@@ -985,10 +1020,10 @@ A second review with working PoCs found that the identities bind less than the f
**Layout binding.**
- A pool entry is referenced at most once within a segment table, and by one table only: two tables may share an entry only when they are the identical table (every reference, size and usable range equal, as for a shared outer volume or a PAR2 file listed by several files). Tables are keyed everywhere (validation, sampling) by `bundle.KeyOf`, a hash of the whole table, never by their first message-id. PoCs P1 (one article referenced 1,999 times as a 1.4 GB file) and P4 (two files sharing a first article, the second never sampled) are regression tests.
-- Interior segments of every table are used whole (`Start == 0`, `End == Size-1`); only the first may start inside its article and only the last may end inside it (`ErrPartialRange`). Real archive slicing in this importer is expressed by extents into outer volumes, never by interior partial ranges of an inline table, so no other layout kind was needed. PoC P2 (100 articles of 700 KB each used for 10 bytes) is a regression test. In the files scheme a declared segment size is bounded by 4 MiB (`MaxFilesSchemeSegment`): posters use 0.25 to 1 MB articles (the importer's own estimates assume about 768 KB) and nothing in the importer or the readers expects more; the nzb scheme keeps the 8 MiB wire cap and is additionally bounded by the user's NZB.
-- Sampling. Every table gets its first and last article and at least one uniformly random interior article (crypto/rand); tables of media files get one per 500 segments, at least three. The cap is 600 articles and 512 MiB (`MaxYencSamples`, `MaxYencBytes`) within 90 seconds (`YencTimeout`). A layout whose tables do not fit at the minimum rate is rejected (`too_many_tables`), never left unsampled. Measured at 768 KB articles: a single 1.1 GB file reads 5 articles (4 MB); a 50-volume release 150 articles (115 MB); a 200-volume release 600 articles (461 MB); 201 volumes are rejected. When the user's NZB is present the first article's yEnc `name=` must equal the subject file name of the NZB entry the table reads (obfuscated names bind too); a disagreement with the local NZB rejects the bundle but is not reported against the publisher.
+- Interior segments of every table are used whole (superseded - see Current rules) (`Start == 0`, `End == Size-1`); only the first may start inside its article and only the last may end inside it (`ErrPartialRange`). Real archive slicing in this importer is expressed by extents into outer volumes, never by interior partial ranges of an inline table, so no other layout kind was needed. PoC P2 (100 articles of 700 KB each used for 10 bytes) is a regression test. In the files scheme a declared segment size is bounded by 4 MiB (`MaxFilesSchemeSegment`): posters use 0.25 to 1 MB articles (the importer's own estimates assume about 768 KB) and nothing in the importer or the readers expects more; the nzb scheme keeps the 8 MiB wire cap and is additionally bounded by the user's NZB.
+- Sampling (superseded - see Current rules). Every table gets its first and last article and at least one uniformly random interior article (crypto/rand); tables of media files get one per 500 segments, at least three. The cap is 600 articles and 512 MiB (`MaxYencSamples`, `MaxYencBytes`) within 90 seconds (`YencTimeout`). A layout whose tables do not fit at the minimum rate is rejected (`too_many_tables`), never left unsampled. Measured at 768 KB articles: a single 1.1 GB file reads 5 articles (4 MB); a 50-volume release 150 articles (115 MB); a 200-volume release 600 articles (461 MB); 201 volumes are rejected. When the user's NZB is present the first article's yEnc `name=` must equal the subject file name of the NZB entry the table reads (obfuscated names bind too); a disagreement with the local NZB rejects the bundle but is not reported against the publisher.
- Forced full check. A pending peer import additionally gets a header sample proportional to its size (1 percent of its articles, at least 20, at most 500, bounded bytes) through an optional `health.PeerHeaderVerifier` installed together with the peer gate (nil: no change; `internal/health` imports nothing from the peer network). A mismatch fails the import and is the only failure attributable to the publisher. This is sampling, not proof.
-- Open network and nested layouts. Archive layouts (outer volumes, extents, AES keys, clips) cannot be bound without reading archive headers. In the open network only plain layouts are accepted (`layout_not_allowed_open`); private networks keep nested layouts, and what a node may serve is unchanged. Open-network bundles for archive releases need archive-header spot checks (planned).
+- Open network and nested layouts (superseded - see Current rules). Archive layouts (outer volumes, extents, AES keys, clips) cannot be bound without reading archive headers. In the open network only plain layouts are accepted (`layout_not_allowed_open`); private networks keep nested layouts, and what a node may serve is unchanged. Open-network bundles for archive releases need archive-header spot checks (planned).
- Structure check. A fragmented MP4 must end in a `moof`+`mdat` pair whose sizes chain to the end of the file or an `mfra` with a valid `mfro` (optionally followed by `free`/`skip`); a non-fragmented MP4 walks up to 256 top-level box headers instead of a 64-read cap; Matroska reads three random interior positions of 1 MiB (three tries each) and requires two consecutive Cluster-interior elements (SimpleBlock, BlockGroup, Timecode, Cluster) in each; MPEG-TS checks eight random positions on the packet grid. The open network accepts only containers with a real check (Matroska/WebM, MP4/MOV, MPEG-TS/M2TS); other media is refused there (`container_not_verifiable`). Still unprovable: bytes inside `mdat`, inside a Matroska frame or between sampled TS packets, and everything between sampled positions. A forger who puts real structure everywhere the check can look, and garbage elsewhere, is not detected by structure alone.
**Trust and fairness.** Corroboration (bundle offers and phase 2 hints, `KnownPrefixes`) keys on the remote IP actually connected to when the peer was last verified (`peernet_peers.connected_ip`, migration 063; /24 for IPv4, /48 for IPv6); a peer with none counts for nothing, so two DNS names for one server are one range. Attesters that rank offers must themselves be known. The fast path hands its requirements (open allowed, corroboration floor) to the fetcher, which filters offers before dialing anyone, so with the switch off no open-network peer is contacted. The serve-side build budget is spent only after the build slot is held and the build starts, and a requester (node ID, public address) can use at most a sixth of it with a burst of one; cache hits and refused requests cost nothing. Only one bundle is verified at a time node-wide (`bundle.VerifyContext`); `MaxDecompressedSize` is 16 MiB (a 50 GB release is about 3.2 MB canonical, so roughly 340,000 segments, about 250 GB at 768 KB per article, still fits comfortably). Decode peak heap fell from 6.5 times the canonical size to about 3.3 times (single exact read buffer, streaming canonical comparison instead of a second full re-encode).
@@ -1006,8 +1041,8 @@ Owner decision: every kind of release is accepted on every network. The second r
- Wrapper check. When the outer table's first segment starts inside its article (an inner volume sliced out of an outer archive), the table is also read raw from the start of its first article; if that article starts a RAR archive, the entry whose data begins at the table's first byte must be stored, as large as the table, and (RAR 5 file encryption record) carry the bundle's IV. RAR 4 keeps its IV derived from the password and cannot be compared. A wrong key or IV shows as garbage where the inner volume's signature should be.
- Bounds. Per scan: 64 blocks, 160 reads, 64 KiB per header, 64 KiB 7z end header; per volume at most 24 distinct articles touched; every touched article is charged (declared size) to the accept-wide budget; at most 64 extents are checked (all of them up to that, else the first and last extent of the main media file plus a crypto/rand choice). No size from the archive is used to allocate; every offset is added with overflow checks; no recursion. Fuzz targets `FuzzVerify` and `FuzzVerifyUnchecked` (checksums off, to reach the parsers below them) run over seed archives built by `archivetest`.
- Outcomes. `Bound`; `Unverifiable` (unknown format, encoded 7z header, encrypted archive headers, budget spent, damaged header chain, read error); `Mismatch`. A mismatch is attributable: `extent_mismatch`, fetch-domain report, negative cache. A missing article is a probe error, never a mismatch.
-- Accepted without header binding, always only with a structural media check through the layout (a generic container is refused, `extent_unverifiable`): disc-image extents; 7z with an encoded end header; encrypted RAR headers (not shareable on the publisher side anyway); volumes whose walk exceeded the budget; every file whose outer volumes are not a RAR or 7z archive. Extents beyond the 64 checked are covered by sampling only. Inline single-archive files (no extents) are accepted on the head and structure checks, as before.
-- Found while studying the producer, not changed: a stored multi-volume RAR or 7z file with no outer encryption is flattened into one inline table with partial ranges at every volume boundary, which `Validate` rejects (`ErrPartialRange`), so such releases are not shareable today; and the yEnc whole-file check (offset zero, table total equals the file total) is only right for tables that cover a posted file from its first to its last article, so a table that is a slice from the middle of a larger posted file is refused by it.
+- Accepted without header binding, always only with a structural media check through the layout (a generic container is refused, `extent_unverifiable`): disc-image extents; 7z with an encoded end header; encrypted RAR headers (not shareable on the publisher side anyway); volumes whose walk exceeded the budget; every file whose outer volumes are not a RAR or 7z archive. Extents beyond the 64 checked are covered by sampling only. Inline single-archive files (no extents) are accepted on the head and structure checks, as before (superseded - see Current rules).
+- Found while studying the producer (superseded - see Current rules): a stored multi-volume RAR or 7z file with no outer encryption is flattened into one inline table with partial ranges at every volume boundary, which `Validate` rejects (`ErrPartialRange`), so such releases are not shareable today; and the yEnc whole-file check (offset zero, table total equals the file total) is only right for tables that cover a posted file from its first to its last article, so a table that is a slice from the middle of a larger posted file is refused by it.
**One budget.** The yEnc header samples and the archive header reads share one budget of declared article bytes (`costBudget`). Order: mandatory samples, archive header checks, then optional samples with what is left.
- Mandatory: the last article of every table (a size lie anywhere in a table shifts every later offset, so the last article's offset and the table total expose a single lie with certainty); for the largest media file's tables also the first article and random interior ones (one per 500, at least three).
diff --git a/frontend/src/components/config/peernet/ActivityCard.tsx b/frontend/src/components/config/peernet/ActivityCard.tsx
index cee4dbb5c..85d45a1aa 100644
--- a/frontend/src/components/config/peernet/ActivityCard.tsx
+++ b/frontend/src/components/config/peernet/ActivityCard.tsx
@@ -4,6 +4,7 @@ import {
backboneHint,
describeBreakdown,
fetchFailLabel,
+ headerBindingLine,
importRejectLabel,
libraryShareLine,
notShareableLabel,
@@ -123,6 +124,7 @@ function BundlesBlock({ status }: { status: PeernetStatus }) {
Awaiting full check: {count(imp?.full_check_passed)} passed,{" "}
{count(imp?.full_check_failed)} failed
+ {headerBindingLine(imp) && {headerBindingLine(imp)}
}
{(imp?.pending ?? 0) > 0 && (
Waiting for a full check: {count(imp?.pending)}
diff --git a/frontend/src/components/config/peernet/peernetStats.ts b/frontend/src/components/config/peernet/peernetStats.ts
index 5a0934263..3fe4709e1 100644
--- a/frontend/src/components/config/peernet/peernetStats.ts
+++ b/frontend/src/components/config/peernet/peernetStats.ts
@@ -1,4 +1,4 @@
-import type { PeernetLibraryStatus } from "../../../types/peernet";
+import type { PeernetBundleImportStatus, PeernetLibraryStatus } from "../../../types/peernet";
// Plain-language labels for the machine keys the status endpoint reports. The
// lookups use Maps so a key such as "__proto__" can never reach a prototype;
@@ -60,6 +60,7 @@ const NOT_SHAREABLE_LABELS: ReadonlyMap = new Map([
["too_large", "Too large to bundle"],
["no_post_date", "No post date known"],
["no_files", "No files in the release"],
+ ["no_media_file", "No media file a receiver would accept"],
["self_check_failed", "The built bundle failed its own check"],
["other", "Other reasons"],
]);
@@ -341,3 +342,21 @@ export function backboneHint(
if (typeof n !== "number" || !Number.isFinite(n) || n <= 0) return null;
return "Health records are not shared because no provider backbone is recognised. Set a backbone (storage group) on your providers, or use a provider host that is in the known list.";
}
+
+/**
+ * One line on how well the files of accepted bundles were tied to their
+ * archive headers at accept time, or null when nothing was checked yet. "Not
+ * tied" parts were accepted on structure alone; they are the ones a full check
+ * matters most for.
+ */
+export function headerBindingLine(imp: PeernetBundleImportStatus | undefined): string | null {
+ if (!imp) return null;
+ const pb = count(imp.parts_header_bound);
+ const pu = count(imp.parts_header_unbound);
+ const eb = count(imp.extents_header_bound);
+ const eu = count(imp.extents_header_unbound);
+ if (pb + pu + eb + eu === 0) return null;
+ const parts = `${pb} ${pb === 1 ? "part" : "parts"} tied to their archive header, ${pu} not tied`;
+ if (eb + eu === 0) return `Archive header checks: ${parts}.`;
+ return `Archive header checks: ${parts}; ${eb} nested ${eb === 1 ? "extent" : "extents"} tied, ${eu} not tied.`;
+}
diff --git a/frontend/src/types/peernet.ts b/frontend/src/types/peernet.ts
index 43130bb88..5b828903f 100644
--- a/frontend/src/types/peernet.ts
+++ b/frontend/src/types/peernet.ts
@@ -134,6 +134,11 @@ export interface PeernetBundleImportStatus {
pending?: number;
/** Pending past the maximum age: failed, never announced. */
stale?: number;
+ /** Parts (and nested-layout extents) of accepted bundles bound to an archive header, and those that could not be. */
+ parts_header_bound?: number;
+ parts_header_unbound?: number;
+ extents_header_bound?: number;
+ extents_header_unbound?: number;
}
export interface PeernetBundlesStatus {
diff --git a/frontend/tests/peernetStats.test.ts b/frontend/tests/peernetStats.test.ts
index 6164d2615..c961cc1ab 100644
--- a/frontend/tests/peernetStats.test.ts
+++ b/frontend/tests/peernetStats.test.ts
@@ -6,6 +6,7 @@ import {
fetchFailLabel,
formatRelativeTime,
hasAnyPublishingLevel,
+ headerBindingLine,
importRejectLabel,
levelNeedsConfirmation,
libraryShareLine,
@@ -85,6 +86,7 @@ const BACKEND_NOT_SHAREABLE_KEYS = [
"too_large",
"no_post_date",
"no_files",
+ "no_media_file",
"self_check_failed",
"other",
];
@@ -434,3 +436,26 @@ describe("backbone hint", () => {
expect(backboneHint(undefined)).toBeNull();
});
});
+
+describe("headerBindingLine", () => {
+ test("says nothing before any part or extent was checked", () => {
+ expect(headerBindingLine(undefined)).toBeNull();
+ expect(headerBindingLine({ parts_header_bound: 0, parts_header_unbound: 0 })).toBeNull();
+ });
+
+ test("reports bound and unbound parts and extents", () => {
+ expect(headerBindingLine({ parts_header_bound: 12, parts_header_unbound: 0 })).toBe(
+ "Archive header checks: 12 parts tied to their archive header, 0 not tied.",
+ );
+ expect(
+ headerBindingLine({
+ parts_header_bound: 3,
+ parts_header_unbound: 1,
+ extents_header_bound: 2,
+ extents_header_unbound: 0,
+ }),
+ ).toBe(
+ "Archive header checks: 3 parts tied to their archive header, 1 not tied; 2 nested extents tied, 0 not tied.",
+ );
+ });
+});
diff --git a/internal/api/peernet_handlers.go b/internal/api/peernet_handlers.go
index 299e2a562..e2a4c6ecc 100644
--- a/internal/api/peernet_handlers.go
+++ b/internal/api/peernet_handlers.go
@@ -201,6 +201,11 @@ type peernetImportDTO struct {
BadBundleReports int64 `json:"bad_bundle_reports"`
Pending int64 `json:"pending"`
Stale int64 `json:"stale"`
+
+ PartsHeaderBound int64 `json:"parts_header_bound"`
+ PartsHeaderUnbound int64 `json:"parts_header_unbound"`
+ ExtentsHeaderBound int64 `json:"extents_header_bound"`
+ ExtentsHeaderUnbound int64 `json:"extents_header_unbound"`
}
func copyCounts(m map[string]int64) map[string]int64 {
@@ -221,7 +226,9 @@ func toBundlesDTO(b service.BundleStatus) peernetBundlesDTO {
Import: peernetImportDTO{Accepted: b.Import.Accepted, AlreadyImported: b.Import.AlreadyImported,
Rejected: copyCounts(b.Import.Rejected), FullCheckPassed: b.Import.FullCheckPassed,
FullCheckFailed: b.Import.FullCheckFailed, Announced: b.Import.Announced, BadBundleReports: b.Import.BadBundleReports,
- Pending: b.Import.Pending, Stale: b.Import.Stale},
+ Pending: b.Import.Pending, Stale: b.Import.Stale,
+ PartsHeaderBound: b.Import.PartsHeaderBound, PartsHeaderUnbound: b.Import.PartsHeaderUnbound,
+ ExtentsHeaderBound: b.Import.ExtentsHeaderBound, ExtentsHeaderUnbound: b.Import.ExtentsHeaderUnbound},
}
}
diff --git a/internal/database/queue_addtoqueue_id_test.go b/internal/database/queue_addtoqueue_id_test.go
new file mode 100644
index 000000000..91c7b59dd
--- /dev/null
+++ b/internal/database/queue_addtoqueue_id_test.go
@@ -0,0 +1,56 @@
+package database
+
+import (
+ "context"
+ "database/sql"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// AddToQueue must leave the id of the row with the item's path in item.ID
+// whether it inserted, updated or (while the row is pending or processing)
+// left the existing row alone; a stale last-insert id or zero would be an
+// invalid foreign key for every later write that refers to the item.
+func TestAddToQueueResolvesTheExistingRowOnEveryDialect(t *testing.T) {
+ backends := map[string]func(t *testing.T) *QueueRepository{
+ "sqlite": func(t *testing.T) *QueueRepository {
+ db, err := sql.Open("sqlite3", ":memory:")
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = db.Close() })
+ db.SetMaxOpenConns(1)
+ setupQueueSchema(t, db)
+ return NewQueueRepository(db, DialectSQLite)
+ },
+ }
+ if hasPostgresDSN() {
+ backends["postgres"] = func(t *testing.T) *QueueRepository {
+ return NewQueueRepository(openPostgresLatest(t), DialectPostgres)
+ }
+ }
+ for name, mk := range backends {
+ t.Run(name, func(t *testing.T) {
+ ctx := context.Background()
+ r := mk(t)
+ first := &ImportQueueItem{NzbPath: "peer://a.nzb", Priority: QueuePriorityNormal, Status: QueueStatusCompleted, MaxRetries: 1}
+ require.NoError(t, r.AddToQueue(ctx, first))
+ require.NotZero(t, first.ID)
+ // Other rows move the last-insert id.
+ for _, p := range []string{"/x/1.nzb", "/x/2.nzb"} {
+ require.NoError(t, r.AddToQueue(ctx, &ImportQueueItem{NzbPath: p, Priority: QueuePriorityNormal, Status: QueueStatusPending, MaxRetries: 1}))
+ }
+ again := &ImportQueueItem{NzbPath: "peer://a.nzb", Priority: QueuePriorityNormal, Status: QueueStatusCompleted, MaxRetries: 1}
+ require.NoError(t, r.AddToQueue(ctx, again))
+ assert.Equal(t, first.ID, again.ID, "an update resolves to the existing row")
+
+ busy := &ImportQueueItem{NzbPath: "/x/1.nzb", Priority: QueuePriorityNormal, Status: QueueStatusPending, MaxRetries: 1}
+ require.NoError(t, r.AddToQueue(ctx, busy)) // left alone while pending
+ require.NoError(t, r.AddToQueue(ctx, &ImportQueueItem{NzbPath: "/x/3.nzb", Priority: QueuePriorityNormal, Status: QueueStatusPending, MaxRetries: 1}))
+ busy2 := &ImportQueueItem{NzbPath: "/x/1.nzb", Priority: QueuePriorityNormal, Status: QueueStatusPending, MaxRetries: 1}
+ require.NoError(t, r.AddToQueue(ctx, busy2))
+ assert.NotZero(t, busy2.ID)
+ assert.Equal(t, busy.ID, busy2.ID, "a skipped upsert resolves to the existing row")
+ })
+ }
+}
diff --git a/internal/database/queue_repository.go b/internal/database/queue_repository.go
index 9d7082e9d..9806fc5d5 100644
--- a/internal/database/queue_repository.go
+++ b/internal/database/queue_repository.go
@@ -232,10 +232,21 @@ func (r *QueueRepository) AddToQueue(ctx context.Context, item *ImportQueueItem)
if err != nil {
return fmt.Errorf("failed to add queue item: %w", err)
}
- item.ID, err = result.LastInsertId()
- if err != nil {
+ // LastInsertId is the connection's previous insert when the upsert
+ // updated an existing row, so it never identifies that row.
+ if _, err = result.LastInsertId(); err != nil {
return fmt.Errorf("failed to get last insert ID: %w", err)
}
+ item.ID = 0
+ }
+ if item.ID == 0 {
+ // The upsert updated (or, on PostgreSQL, left alone) an existing row:
+ // the row is the one with this path.
+ var id int64
+ if err := r.db.QueryRowContext(ctx, `SELECT id FROM import_queue WHERE nzb_path = ?`, item.NzbPath).Scan(&id); err != nil {
+ return fmt.Errorf("failed to resolve queue item %q after upsert: %w", item.NzbPath, err)
+ }
+ item.ID = id
}
item.CreatedAt = time.Now()
diff --git a/internal/importer/peer_bundle_realpipeline_test.go b/internal/importer/peer_bundle_realpipeline_test.go
new file mode 100644
index 000000000..66b1be259
--- /dev/null
+++ b/internal/importer/peer_bundle_realpipeline_test.go
@@ -0,0 +1,244 @@
+package importer
+
+import (
+ "context"
+ "crypto/aes"
+ "crypto/cipher"
+ "crypto/pbkdf2"
+ "crypto/sha256"
+ "fmt"
+ "io"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/javi11/nntppool/v5"
+ "github.com/kipsilabs/altmount/internal/config"
+ "github.com/kipsilabs/altmount/internal/database"
+ "github.com/kipsilabs/altmount/internal/metadata"
+ metapb "github.com/kipsilabs/altmount/internal/metadata/proto"
+ "github.com/kipsilabs/altmount/internal/peernet/archivehdr/archivetest"
+ "github.com/kipsilabs/altmount/internal/peernet/bundle"
+ "github.com/kipsilabs/altmount/internal/peernet/bundleimport"
+ "github.com/kipsilabs/altmount/internal/peernet/bundlex"
+ "github.com/kipsilabs/altmount/internal/peernet/contentid"
+ "github.com/kipsilabs/altmount/internal/testsupport/fakepool"
+ "github.com/kipsilabs/altmount/internal/testsupport/mediagen"
+ "github.com/kipsilabs/altmount/internal/testsupport/nzbbuild"
+ "github.com/stretchr/testify/require"
+)
+
+// notShareable lists release types a publisher is allowed to refuse, with the
+// reason. A type that is not here must build, verify, be accepted by a
+// receiver and read back byte-identically.
+var notShareable = map[string]string{}
+
+type rpProber struct{}
+
+func (rpProber) ProbeRelease(context.Context, []bundleimport.ProbeFile) error { return nil }
+
+type rpAnnouncer struct{}
+
+func (rpAnnouncer) AnnounceVerifiedImport(context.Context, bundleimport.Announcement) error {
+ return nil
+}
+
+type rpReporter struct{}
+
+func (rpReporter) ReportBadBundle(context.Context, bundleimport.Origin, string) {}
+func (rpReporter) ReportVerifiedBundle(context.Context, bundleimport.Origin, string) {}
+func (rpReporter) ReportFailedBundle(context.Context, string, string) {}
+
+type rpOrigins struct{ fromNZB bool }
+
+func (o rpOrigins) FromNZB(context.Context, string) (bool, error) { return o.fromNZB, nil }
+
+type rpLibrary struct{ paths []string }
+
+func (l rpLibrary) ReleaseFiles(context.Context, string) (int, []string, error) {
+ return database.PeernetSchemeNZB, l.paths, nil
+}
+
+// rpRelease is one release type of the matrix: the posted NZB files and the
+// payload (the media file as the library will show it).
+type rpRelease struct {
+ name string
+ payload string // library file name of the payload
+ data []byte
+ password string
+ build func(post func(name string, data []byte) nzbbuild.File) []nzbbuild.File
+}
+
+const (
+ rpSeg = 40_000
+ rpPayload = 500_000
+)
+
+func rpRar5Volumes(name string, pl []byte, n int) func(func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ return func(post func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ vol := (len(pl) + n - 1) / n
+ var out []nzbbuild.File
+ for i := 0; i < n; i++ {
+ lo, hi := i*vol, min(len(pl), (i+1)*vol)
+ b := archivetest.RAR5Volume(i, archivetest.File{Name: name, Data: pl[lo:hi], SplitBefore: i > 0, SplitAfter: i < n-1, UnpackedSize: int64(len(pl))})
+ out = append(out, post(fmt.Sprintf("Release.part%02d.rar", i+1), b.Data))
+ }
+ return out
+ }
+}
+
+func rpRar4Volumes(name string, pl []byte, n int) func(func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ return func(post func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ vol := (len(pl) + n - 1) / n
+ var out []nzbbuild.File
+ for i := 0; i < n; i++ {
+ lo, hi := i*vol, min(len(pl), (i+1)*vol)
+ b := archivetest.RAR4Volume(i, i == n-1, archivetest.File{Name: name, Data: pl[lo:hi], SplitBefore: i > 0, SplitAfter: i < n-1, UnpackedSize: int64(len(pl))})
+ out = append(out, post(fmt.Sprintf("Release.part%02d.rar", i+1), b.Data))
+ }
+ return out
+ }
+}
+
+func rpEncrypted(name string, pl []byte, pw string) func(func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ return func(post func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ salt := make([]byte, 16)
+ iv := []byte("0123456789abcdef")
+ key, err := pbkdf2.Key(sha256.New, pw, salt, 1<<15, 32)
+ if err != nil {
+ panic(err)
+ }
+ blk, _ := aes.NewCipher(key)
+ padded := append([]byte{}, pl...)
+ for len(padded)%16 != 0 {
+ padded = append(padded, 0)
+ }
+ cipher.NewCBCEncrypter(blk, iv).CryptBlocks(padded, padded)
+ b := archivetest.RAR5(archivetest.File{Name: name, Data: padded, IV: iv, UnpackedSize: int64(len(pl))})
+ return []nzbbuild.File{post("Release.rar", b.Data)}
+ }
+}
+
+func rpReleases() []rpRelease {
+ mkv := mediagen.MKV(rpPayload)
+ avi := mediagen.AVI(rpPayload)
+ plain := func(n string, d []byte) func(func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ return func(post func(string, []byte) nzbbuild.File) []nzbbuild.File { return []nzbbuild.File{post(n, d)} }
+ }
+ return []rpRelease{
+ {name: "plain mkv", payload: "Movie.2024.mkv", data: mkv, build: plain("Movie.2024.mkv", mkv)},
+ {name: "plain avi (generic container)", payload: "Movie.2024.avi", data: avi, build: plain("Movie.2024.avi", avi)},
+ {name: "rar5 single volume stored", payload: "Movie.2024.mkv", data: mkv, build: func(post func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ return []nzbbuild.File{post("Release.rar", archivetest.RAR5(archivetest.File{Name: "Movie.2024.mkv", Data: mkv}).Data)}
+ }},
+ {name: "rar5 multi volume stored (4)", payload: "Movie.2024.mkv", data: mkv, build: rpRar5Volumes("Movie.2024.mkv", mkv, 4)},
+ {name: "rar4 single volume stored", payload: "Movie.2024.mkv", data: mkv, build: func(post func(string, []byte) nzbbuild.File) []nzbbuild.File {
+ return []nzbbuild.File{post("Release.rar", archivetest.RAR4(archivetest.File{Name: "Movie.2024.mkv", Data: mkv}).Data)}
+ }},
+ {name: "rar4 multi volume stored (4)", payload: "Movie.2024.mkv", data: mkv, build: rpRar4Volumes("Movie.2024.mkv", mkv, 4)},
+ {name: "rar5 encrypted, password in the NZB", payload: "Movie.2024.mkv", data: mkv, password: "fixture-pw-123", build: rpEncrypted("Movie.2024.mkv", mkv, "fixture-pw-123")},
+ }
+}
+
+func rpVirtualPaths(t *testing.T, root string) []string {
+ t.Helper()
+ var out []string
+ require.NoError(t, filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+ if err == nil && !d.IsDir() && strings.HasSuffix(p, ".meta") {
+ rel, _ := filepath.Rel(root, strings.TrimSuffix(p, ".meta"))
+ out = append(out, filepath.ToSlash(rel))
+ }
+ return err
+ }))
+ return out
+}
+
+func TestRealPipelineEveryReleaseTypeIsShareableAndAccepted(t *testing.T) {
+ for _, rel := range rpReleases() {
+ t.Run(rel.name, func(t *testing.T) {
+ ctx := context.Background()
+ // Publisher: the real importer over the fake pool.
+ env := newBatteryEnv(t)
+ post := func(name string, data []byte) nzbbuild.File {
+ var segs []nzbbuild.Segment
+ total := (len(data) + rpSeg - 1) / rpSeg
+ for i := 0; i < total; i++ {
+ lo, hi := i*rpSeg, min(len(data), (i+1)*rpSeg)
+ id := fmt.Sprintf("%s-%s-%03d@rp.test", strings.ReplaceAll(rel.name, " ", ""), name, i+1)
+ env.client.SetBehavior(id, fakepool.SegmentBehavior{Bytes: data[lo:hi], YEnc: nntppool.YEncMeta{
+ FileName: name, FileSize: int64(len(data)), Part: int64(i + 1), Total: int64(total), PartBegin: int64(lo), PartSize: int64(hi - lo)}})
+ segs = append(segs, nzbbuild.Segment{ID: id, Bytes: hi - lo})
+ }
+ return nzbbuild.File{Subject: name, Segments: segs}
+ }
+ nzb := nzbbuild.Build(rel.build(post)...)
+ if rel.password != "" {
+ nzb.Meta = map[string]string{"password": rel.password}
+ }
+ _, _, err := env.runImport(nzb, "Release.Name.2024.1080p")
+ require.NoError(t, err, "the real importer must import the fixture")
+ paths := rpVirtualPaths(t, env.metaRoot)
+ require.NotEmpty(t, paths)
+
+ src := bundlex.NewMetadataSource(env.svc, rpOrigins{fromNZB: rel.password != ""})
+ lists, err := src.NZBSegmentIDs(paths[0])
+ require.NoError(t, err)
+ store := &metapb.NzbStore{}
+ for _, l := range lists {
+ fe := &metapb.NzbFileEntry{}
+ for _, id := range l {
+ fe.Segments = append(fe.Segments, &metapb.NzbSeg{Id: id})
+ }
+ store.Files = append(store.Files, fe)
+ }
+ rid, err := contentid.ReleaseID(store)
+ require.NoError(t, err)
+
+ builder := bundlex.NewBuilder(bundlex.BuilderConfig{Source: src, Library: rpLibrary{paths: paths}, Patches: bundlex.NoPatches{}})
+ built, err := builder.Build(ctx, rid.String(), "Release Name 2024 1080p", time.Unix(1_700_000_000, 0))
+ if reason, ok := notShareable[rel.name]; ok {
+ require.Error(t, err, "documented not shareable: %s", reason)
+ return
+ }
+ require.NoError(t, err, "a release the importer produced must be shareable")
+ t.Logf("MATRIX %-40s built=yes scheme=%d wire=%d bytes", rel.name, built.Scheme, len(built.Wire))
+ bd, err := bundle.Verify(built.Wire, built.ID, rid)
+ require.NoError(t, err)
+
+ // Receiver: real accept over the same fake pool, real stager.
+ lib := metadata.NewMetadataService(t.TempDir())
+ db, err := database.NewDB(database.Config{DatabasePath: filepath.Join(t.TempDir(), "db")})
+ require.NoError(t, err)
+ t.Cleanup(func() { _ = db.Close() })
+ cfg := config.DefaultConfig()
+ pm := processorTestPoolManager{client: env.client}
+ stager := bundleimport.NewFSStager(bundleimport.FSStagerDeps{Pool: pm, Config: func() *config.Config { return cfg }, Root: t.TempDir()})
+ imp := bundleimport.New(bundleimport.Config{
+ Library: lib, Stager: stager, Prober: rpProber{}, Headers: bundleimport.NewPoolHeaderReader(pm),
+ State: database.NewPeernetBundleImportRepository(db.Connection(), database.DialectSQLite), Announcer: rpAnnouncer{}, Reporter: rpReporter{},
+ HealthEnabled: func() bool { return true },
+ })
+ res, err := imp.Accept(ctx, bundleimport.Request{Bundle: bd, Target: bundleimport.Target{Dir: "movies/Release.Name.2024.1080p", Category: "movies"}})
+ require.NoError(t, err, "a receiver must accept what the publisher announced")
+ require.Len(t, res.Paths, 1)
+ require.True(t, strings.HasSuffix(res.Paths[0], "/"+rel.payload), "accepted file %q, want name %q", res.Paths[0], rel.payload)
+
+ stage, err := stager.NewStaging()
+ require.NoError(t, err)
+ defer stage.Close()
+ m, err := lib.ReadFileMetadata(res.Paths[0])
+ require.NoError(t, err)
+ require.NoError(t, stage.WriteFileMetadata(res.Paths[0], m))
+ f, err := stage.Opener().OpenFile(ctx, res.Paths[0], os.O_RDONLY, 0)
+ require.NoError(t, err)
+ defer f.Close()
+ got, err := io.ReadAll(f)
+ require.NoError(t, err)
+ require.Equal(t, sha256.Sum256(rel.data), sha256.Sum256(got), "accepted file bytes differ from the payload")
+ t.Logf("MATRIX %-40s accepted=yes bytes-identical=yes size=%d", rel.name, len(got))
+ })
+ }
+}
diff --git a/internal/importer/peer_bundle_service_test.go b/internal/importer/peer_bundle_service_test.go
index 4a8627335..628d93b0e 100644
--- a/internal/importer/peer_bundle_service_test.go
+++ b/internal/importer/peer_bundle_service_test.go
@@ -238,3 +238,26 @@ func TestPeerImportDirRefusesTitlesThatEscapeTheCategoryFolder(t *testing.T) {
}
assert.Equal(t, catDir+"/Some Movie", s.PeerImportDir("Movies", "Some Movie.mkv"))
}
+
+// A second direct import of the same release finds its queue row completed,
+// so the upsert changes nothing; the import must still be tied to that row
+// (not to a stale or zero id) and its history entry must be written.
+func TestRecordPeerImportTwiceKeepsTheQueueReferenceValid(t *testing.T) {
+ ctx := context.Background()
+ s := newPeerService(t)
+ rec := PeerImportRecord{Title: "My Release", Category: "movies", ResultingPath: "movies/My Release",
+ WrittenPaths: []string{"movies/My Release/a.mkv"}, TotalSize: 99, ReleaseID: "abcdef0123456789abcdef"}
+ first, err := s.RecordPeerImport(ctx, rec)
+ require.NoError(t, err)
+ // Unrelated inserts on the connection make a stale last-insert id differ.
+ for i := 0; i < 3; i++ {
+ addItem(t, s, false)
+ }
+ second, err := s.RecordPeerImport(ctx, rec)
+ require.NoError(t, err)
+ assert.Equal(t, first, second, "the same release resolves to the same queue row")
+ var n int
+ require.NoError(t, s.database.Connection().QueryRowContext(ctx,
+ `SELECT COUNT(*) FROM import_history WHERE nzb_id = ?`, first).Scan(&n))
+ assert.GreaterOrEqual(t, n, 1, "history written against the queue row")
+}
diff --git a/internal/peernet/archivehdr/archivetest/archivetest.go b/internal/peernet/archivehdr/archivetest/archivetest.go
index 68c43ffda..76a13a053 100644
--- a/internal/peernet/archivehdr/archivetest/archivetest.go
+++ b/internal/peernet/archivehdr/archivetest/archivetest.go
@@ -129,7 +129,16 @@ func rar5(volume int, files []File) Built {
}
// RAR4 builds a RAR 4 archive.
-func RAR4(files ...File) Built {
+func RAR4(files ...File) Built { return rar4(-1, false, files) }
+
+// RAR4Volume builds volume (0-based) of a multi-volume RAR 4 set, one that a
+// real RAR reader follows: the main header carries the volume and
+// new-numbering flags (and first-volume on volume 0), and the end-of-archive
+// block says whether more volumes follow. A file continued into the next
+// volume must set SplitAfter, one continued from the previous SplitBefore.
+func RAR4Volume(volume int, last bool, files ...File) Built { return rar4(volume, last, files) }
+
+func rar4(volume int, last bool, files []File) Built {
var buf bytes.Buffer
buf.Write([]byte{'R', 'a', 'r', '!', 0x1A, 0x07, 0x00})
header := func(kind byte, flags uint16, body []byte) {
@@ -141,7 +150,14 @@ func RAR4(files ...File) Built {
binary.LittleEndian.PutUint16(data[:2], uint16(crc32.ChecksumIEEE(data[2:])))
buf.Write(data)
}
- header(0x73, 0, make([]byte, 6))
+ mainFlags := uint16(0)
+ if volume >= 0 {
+ mainFlags = 0x0001 | 0x0010
+ if volume == 0 {
+ mainFlags |= 0x0100
+ }
+ }
+ header(0x73, mainFlags, make([]byte, 6))
var offsets []int64
for _, f := range files {
body := make([]byte, 25)
@@ -173,7 +189,11 @@ func RAR4(files ...File) Built {
offsets = append(offsets, int64(buf.Len()))
buf.Write(f.Data)
}
- header(0x7B, 0, nil)
+ endFlags := uint16(0)
+ if volume >= 0 && !last {
+ endFlags = 0x0001
+ }
+ header(0x7B, endFlags, nil)
return Built{Data: buf.Bytes(), DataOffsets: offsets}
}
diff --git a/internal/peernet/archivehdr/archivetest/rar4volume_test.go b/internal/peernet/archivehdr/archivetest/rar4volume_test.go
new file mode 100644
index 000000000..443209b0e
--- /dev/null
+++ b/internal/peernet/archivehdr/archivetest/rar4volume_test.go
@@ -0,0 +1,48 @@
+package archivetest_test
+
+import (
+ "encoding/binary"
+ "testing"
+
+ "github.com/kipsilabs/altmount/internal/peernet/archivehdr/archivetest"
+)
+
+// flagsOf returns the flags of the block of the given type in a RAR 4 volume
+// (headers only; the test volumes carry little data).
+func flagsOf(t *testing.T, data []byte, kind byte) uint16 {
+ t.Helper()
+ for pos := 7; pos+7 <= len(data); {
+ size := int(binary.LittleEndian.Uint16(data[pos+5 : pos+7]))
+ flags := binary.LittleEndian.Uint16(data[pos+3 : pos+5])
+ if data[pos+2] == kind {
+ return flags
+ }
+ if flags&0x8000 != 0 && kind != data[pos+2] {
+ size += int(binary.LittleEndian.Uint32(data[pos+7 : pos+11]))
+ }
+ pos += size
+ }
+ t.Fatalf("no block of type %#x", kind)
+ return 0
+}
+
+// A multi-volume RAR 4 set has the volume and new-numbering flags on the
+// main header (first-volume on volume 0) and "more volumes follow" on every
+// end block but the last; the real RAR reader needs all of them.
+func TestRAR4VolumeFlags(t *testing.T) {
+ f := archivetest.File{Name: "a.mkv", Data: []byte("x"), SplitAfter: true, UnpackedSize: 2}
+ first := archivetest.RAR4Volume(0, false, f)
+ if got := flagsOf(t, first.Data, 0x73); got != 0x0111 {
+ t.Fatalf("first volume main flags %#x", got)
+ }
+ if got := flagsOf(t, first.Data, 0x7B); got != 0x0001 {
+ t.Fatalf("first volume end flags %#x", got)
+ }
+ last := archivetest.RAR4Volume(1, true, archivetest.File{Name: "a.mkv", Data: []byte("y"), SplitBefore: true, UnpackedSize: 2})
+ if got := flagsOf(t, last.Data, 0x73); got != 0x0011 {
+ t.Fatalf("last volume main flags %#x", got)
+ }
+ if got := flagsOf(t, last.Data, 0x7B); got != 0 {
+ t.Fatalf("last volume end flags %#x", got)
+ }
+}
diff --git a/internal/peernet/bridge/bundleimport.go b/internal/peernet/bridge/bundleimport.go
index 695fe59ce..35aa2b517 100644
--- a/internal/peernet/bridge/bundleimport.go
+++ b/internal/peernet/bridge/bundleimport.go
@@ -313,6 +313,15 @@ func (r bundleReporter) ReportBadBundle(_ context.Context, o bundleimport.Origin
r.s.NoteBundleFullCheck(o.NetworkID, o.PublisherID, false)
}
+// RefundFetch gives the publisher's fetch budget back when the service can.
+func (r bundleReporter) RefundFetch(_ context.Context, o bundleimport.Origin) {
+ if rf, ok := r.s.(interface {
+ RefundBundleFetch(networkID, publisherID string)
+ }); ok {
+ rf.RefundBundleFetch(o.NetworkID, o.PublisherID)
+ }
+}
+
// ReportFailedBundle puts a layout that failed local verification on the
// negative cache when the service keeps one.
func (r bundleReporter) ReportFailedBundle(_ context.Context, bundleID, releaseID string) {
diff --git a/internal/peernet/bundle/limits.go b/internal/peernet/bundle/limits.go
index b321a3c26..7feebc134 100644
--- a/internal/peernet/bundle/limits.go
+++ b/internal/peernet/bundle/limits.go
@@ -22,6 +22,20 @@ const (
// segment table (main files, shared outer sources and PAR2 files), and
// separately the message-ID pool.
MaxSegments = 500_000
+ // HeaderSlack is the most a part after the first may start into its first
+ // article. What precedes a continuation volume's data is the archive
+ // signature, the main header and one file header: a few hundred bytes for
+ // RAR 4, RAR 5 and 7z, a few KiB with a 2 KiB name and the extra records
+ // (times, hash, encryption). 64 KiB leaves a wide margin and still makes a
+ // 700 KB article give up at most about 9 percent to headers.
+ HeaderSlack = 64 << 10
+ // TrailerSlack is the most a part before the last may stop short of the end
+ // of its last article. A volume ends with an end-of-archive block (tens of
+ // bytes); a RAR recovery record (-rr) stored in the volume can be larger,
+ // up to about ten percent of it, so the bound is the larger of TrailerSlack
+ // and 1/TrailerShareDivisor of the part's articles (12.5 percent).
+ TrailerSlack = 64 << 10
+ TrailerShareDivisor = 8
// MaxFiles caps the files of one bundle.
MaxFiles = 10_000
// MaxReleaseFiles caps the NZB file entries of the "nzb" scheme release
diff --git a/internal/peernet/bundle/poc_regress_test.go b/internal/peernet/bundle/poc_regress_test.go
new file mode 100644
index 000000000..5356d8fce
--- /dev/null
+++ b/internal/peernet/bundle/poc_regress_test.go
@@ -0,0 +1,169 @@
+package bundle_test
+
+import (
+ "errors"
+ "fmt"
+ "testing"
+ "time"
+
+ metapb "github.com/kipsilabs/altmount/internal/metadata/proto"
+ "github.com/kipsilabs/altmount/internal/peernet/bundle"
+ "github.com/kipsilabs/altmount/internal/peernet/contentid"
+)
+
+func mid(i int) string { return fmt.Sprintf("part%d@poc.example", i) }
+
+// seal fills the identities and returns the Verify result.
+func seal(t *testing.T, b *bundle.Bundle) error {
+ b.Title, b.PostedUnix = "Poc.Release", 1700000000
+ var sum int64
+ for i := range b.Files {
+ b.Files[i].FileID = contentid.ID{byte(i + 1)}
+ sum += b.Files[i].Size
+ }
+ b.TotalSize = sum
+ // provisional ReleaseID so ToLocal's Validate passes
+ b.ReleaseID = contentid.ID{9}
+ loc, err := bundle.ToLocal(b, time.Now())
+ if err != nil {
+ return fmt.Errorf("validate: %w", err)
+ }
+ ids := []contentid.ID{}
+ for i, lf := range loc.Files {
+ id, err := contentid.FileID(lf.Meta)
+ if err != nil {
+ return err
+ }
+ b.Files[i].FileID = id
+ ids = append(ids, id)
+ }
+ sortFiles(b)
+ ids = ids[:0]
+ for i := range b.Files {
+ ids = append(ids, b.Files[i].FileID)
+ }
+ if b.Scheme == bundle.SchemeFiles {
+ b.ReleaseID = bundle.FilesReleaseID(ids)
+ } else {
+ store := &metapb.NzbStore{}
+ for _, f := range loc.NZBFiles {
+ fe := &metapb.NzbFileEntry{}
+ for _, id := range f {
+ fe.Segments = append(fe.Segments, &metapb.NzbSeg{Id: id})
+ }
+ store.Files = append(store.Files, fe)
+ }
+ b.ReleaseID, err = contentid.ReleaseID(store)
+ if err != nil {
+ return err
+ }
+ }
+ wire, id, err := bundle.Pack(b)
+ if err != nil {
+ return fmt.Errorf("pack: %w", err)
+ }
+ _, err = bundle.Verify(wire, id, b.ReleaseID)
+ return err
+}
+
+func sortFiles(b *bundle.Bundle) {
+ for i := 0; i < len(b.Files); i++ {
+ for j := i + 1; j < len(b.Files); j++ {
+ if string(b.Files[j].FileID[:]) < string(b.Files[i].FileID[:]) {
+ b.Files[i], b.Files[j] = b.Files[j], b.Files[i]
+ }
+ }
+ }
+}
+
+// Q1: files scheme, a file made of 100 one-segment parts that each take 10
+// bytes from the middle of a 700 KB article. Header binding at accept would
+// be optional for the attacker, so the format itself must refuse it: a part
+// that is neither the first nor the last must read nearly all of its articles.
+func TestQ1FilesSchemeSingleSegmentPartsAreRefused(t *testing.T) {
+ n := 100
+ pool := []string{}
+ segs := []bundle.Segment{}
+ parts := []int{}
+ for i := 0; i < n; i++ {
+ pool = append(pool, mid(i))
+ segs = append(segs, bundle.Segment{Ref: int64(i), Size: 700000, Start: 1000, End: 1009})
+ parts = append(parts, 1)
+ }
+ b := &bundle.Bundle{Scheme: bundle.SchemeFiles, Pool: pool,
+ Files: []bundle.File{{Name: "Movie.mkv", Size: int64(n) * 10, Segments: segs, Parts: parts}}}
+ if err := seal(t, b); !errors.Is(err, bundle.ErrPartialRange) {
+ t.Fatalf("want ErrPartialRange, got %v", err)
+ }
+}
+
+// Q2: the same 100 windows as ONE part must stay refused.
+func TestQ2OnePartOfPartialSegmentsIsRefused(t *testing.T) {
+ n := 100
+ pool := []string{}
+ segs := []bundle.Segment{}
+ for i := 0; i < n; i++ {
+ pool = append(pool, mid(i))
+ segs = append(segs, bundle.Segment{Ref: int64(i), Size: 700000, Start: 1000, End: 1009})
+ }
+ b := &bundle.Bundle{Scheme: bundle.SchemeFiles, Pool: pool,
+ Files: []bundle.File{{Name: "Movie.mkv", Size: int64(n) * 10, Segments: segs, Parts: []int{n}}}}
+ if err := seal(t, b); err == nil {
+ t.Fatal("a single part of 100 partial segments must be refused")
+ }
+}
+
+// Q3: nzb scheme, 100 entries; the file takes one arbitrary 10-byte window of
+// one article of each, in reverse NZB order.
+func TestQ3NZBReorderedWindowsAreRefused(t *testing.T) {
+ entries := 100
+ pool := []string{}
+ rel := []int{}
+ for e := 0; e < entries; e++ {
+ rel = append(rel, 3)
+ for k := 0; k < 3; k++ {
+ pool = append(pool, mid(e*3+k))
+ }
+ }
+ segs := []bundle.Segment{}
+ parts := []int{}
+ for e := entries - 1; e >= 0; e-- {
+ segs = append(segs, bundle.Segment{Ref: int64(e*3 + 1), Size: 700000, Start: 5, End: 14})
+ parts = append(parts, 1)
+ }
+ b := &bundle.Bundle{Scheme: bundle.SchemeNZB, Pool: pool, ReleaseFiles: rel,
+ Files: []bundle.File{{Name: "Movie.mkv", Size: int64(entries) * 10, Segments: segs, Parts: parts}}}
+ if err := seal(t, b); !errors.Is(err, bundle.ErrPartialRange) {
+ t.Fatalf("want ErrPartialRange, got %v", err)
+ }
+}
+
+// An honest stored multi-volume layout still validates: each volume is read
+// from just after its archive headers to just before its end block.
+func TestHonestVolumePartsStillValidate(t *testing.T) {
+ const art = 700000
+ pool := []string{}
+ segs := []bundle.Segment{}
+ parts := []int{}
+ var size int64
+ for v := 0; v < 4; v++ {
+ for k := 0; k < 3; k++ {
+ ref := int64(len(pool))
+ pool = append(pool, mid(int(ref)))
+ s := bundle.Segment{Ref: ref, Size: art, Start: 0, End: art - 1}
+ if k == 0 {
+ s.Start = 300 // headers of the volume
+ }
+ if k == 2 {
+ s.End = art - 1 - 24 // end-of-archive block
+ }
+ segs = append(segs, s)
+ size += s.End - s.Start + 1
+ }
+ parts = append(parts, 3)
+ }
+ b := &bundle.Bundle{Scheme: bundle.SchemeFiles, Pool: pool, Files: []bundle.File{{Name: "Movie.mkv", Size: size, Segments: segs, Parts: parts}}}
+ if err := seal(t, b); err != nil {
+ t.Fatalf("an honest layout must validate: %v", err)
+ }
+}
diff --git a/internal/peernet/bundle/validate.go b/internal/peernet/bundle/validate.go
index e4fa65cd8..6e45735ef 100644
--- a/internal/peernet/bundle/validate.go
+++ b/internal/peernet/bundle/validate.go
@@ -488,7 +488,10 @@ func (v *validator) checkSlicedTable(f *File) (int64, error) {
seen := make(map[int64]struct{}, len(segs))
var covered int64
at := 0
- for _, n := range f.Parts {
+ for pi, n := range f.Parts {
+ if err := checkPartSlack(segs[at:at+n], pi, len(f.Parts)); err != nil {
+ return 0, err
+ }
for k := 0; k < n; k++ {
i, s := at+k, segs[at+k]
first, last := k == 0, k == n-1
@@ -509,6 +512,36 @@ func (v *validator) checkSlicedTable(f *File) (int64, error) {
return covered, nil
}
+// checkPartSlack applies the deterministic part rules: a part of an archive
+// volume set is read from just after the volume's headers to just before its
+// end block, so, apart from the two ends of the file, no part may begin deep
+// inside its first article or stop long before the end of its last one. A
+// part that did could be a sliver of a large article, which no header check
+// can be required to notice. The first part may begin deep (a file stored
+// after another one in the same volume) and the last may end deep (another
+// file follows it); every other boundary is bounded by HeaderSlack and
+// TrailerSlack (see their documentation).
+func checkPartSlack(part []Segment, pi, nParts int) error {
+ if nParts < 2 {
+ return nil
+ }
+ first, last := part[0], part[len(part)-1]
+ if pi > 0 && first.Start > HeaderSlack {
+ return fmt.Errorf("%w: part %d starts %d bytes into its first article", ErrPartialRange, pi, first.Start)
+ }
+ if pi < nParts-1 {
+ var articles int64
+ for _, s := range part {
+ articles += s.Size
+ }
+ tail := last.Size - 1 - last.End
+ if tail > max(TrailerSlack, articles/TrailerShareDivisor) {
+ return fmt.Errorf("%w: part %d stops %d bytes before the end of its last article", ErrPartialRange, pi, tail)
+ }
+ }
+ return nil
+}
+
// shareArticle records the use of a pool entry by a segment of a part and
// refuses every use the boundary exception does not cover.
func (v *validator) shareArticle(i int, s Segment, boundary bool) error {
diff --git a/internal/peernet/bundleimport/accept.go b/internal/peernet/bundleimport/accept.go
index be9099700..b69d34997 100644
--- a/internal/peernet/bundleimport/accept.go
+++ b/internal/peernet/bundleimport/accept.go
@@ -59,6 +59,16 @@ func (i *Importer) rejected(r Reason, err error) (*Result, error) {
// media probe through the layout in a hidden staging area, and only then the
// local write in pending state. Any rejection leaves nothing behind.
func (i *Importer) Accept(ctx context.Context, req Request) (*Result, error) {
+ res, err := i.accept(ctx, req)
+ if err != nil && localPolicyReason(ReasonOf(err)) && req.Origin.PublisherID != "" {
+ if rf, ok := i.cfg.Reporter.(FetchRefunder); ok {
+ rf.RefundFetch(ctx, req.Origin)
+ }
+ }
+ return res, err
+}
+
+func (i *Importer) accept(ctx context.Context, req Request) (*Result, error) {
ctx, cancel := context.WithTimeout(ctx, i.cfg.Policy.AcceptTimeout)
defer cancel()
b := req.Bundle
@@ -175,7 +185,13 @@ func (i *Importer) Accept(ctx context.Context, req Request) (*Result, error) {
if err != nil {
return i.rejected(ReasonWriteFailed, err)
}
- i.cnt.add(func(c *counters) { c.accepted++ })
+ i.cnt.add(func(c *counters) {
+ c.accepted++
+ c.partsBound += uint64(res.Cost.PartsBound)
+ c.partsUnbound += uint64(res.Cost.PartsUnbound)
+ c.extBound += uint64(res.Cost.ExtentsBound)
+ c.extUnbound += uint64(res.Cost.ExtentsUnbound)
+ })
return res, nil
}
diff --git a/internal/peernet/bundleimport/accept_test.go b/internal/peernet/bundleimport/accept_test.go
index 25a757d91..98d312fc2 100644
--- a/internal/peernet/bundleimport/accept_test.go
+++ b/internal/peernet/bundleimport/accept_test.go
@@ -49,7 +49,11 @@ func (a *fakeAnnouncer) AnnounceVerifiedImport(_ context.Context, an Announcemen
return nil
}
-type fakeReporter struct{ got, verified, failedBundles []string }
+type fakeReporter struct{ got, verified, failedBundles, refunds []string }
+
+func (r *fakeReporter) RefundFetch(_ context.Context, o Origin) {
+ r.refunds = append(r.refunds, o.PublisherID)
+}
func (r *fakeReporter) ReportFailedBundle(_ context.Context, bundleID, release string) {
r.failedBundles = append(r.failedBundles, bundleID+"/"+release)
diff --git a/internal/peernet/bundleimport/filepolicy.go b/internal/peernet/bundleimport/filepolicy.go
index 45171a7c1..790dee5b7 100644
--- a/internal/peernet/bundleimport/filepolicy.go
+++ b/internal/peernet/bundleimport/filepolicy.go
@@ -55,7 +55,13 @@ func (i *Importer) selectFiles(b *bundle.Bundle) ([]int, Reason, error) {
if i.cfg.AllowedExtensions != nil {
allowed = i.cfg.AllowedExtensions()
}
- pol := i.cfg.Policy
+ return SelectFiles(b, allowed, i.cfg.Policy)
+}
+
+// SelectFiles is the receiver's pure file policy: it needs no network or
+// state, so a publisher can run it before announcing a bundle that every
+// receiver would certainly refuse.
+func SelectFiles(b *bundle.Bundle, allowed []string, pol Policy) ([]int, Reason, error) {
var keep []int
media, sidecars := 0, 0
for n := range b.Files {
diff --git a/internal/peernet/bundleimport/importer.go b/internal/peernet/bundleimport/importer.go
index 670f10853..d785920e0 100644
--- a/internal/peernet/bundleimport/importer.go
+++ b/internal/peernet/bundleimport/importer.go
@@ -2,14 +2,18 @@ package bundleimport
import (
"errors"
+ "fmt"
"golang.org/x/sync/singleflight"
+ "strings"
"sync"
"time"
)
-// Config holds the collaborators of an Importer. Fetcher, Recorder, Announcer
-// and Reporter may be nil: the matching step is then skipped (no fetch, no
-// direct import, nothing announced, nothing reported).
+// Config holds the collaborators of an Importer. Library, Stager, Prober,
+// State, Headers and HealthEnabled are required (see Missing; NewChecked
+// refuses a Config without them). Index, Fetcher, Recorder, Announcer and
+// Reporter may be nil: the matching step is then skipped (nothing indexed, no
+// fetch, no direct import, nothing announced, nothing reported).
type Config struct {
Library Library
Stager Stager
@@ -35,11 +39,10 @@ type Config struct {
Now func() time.Time
}
-// Missing lists the collaborators the accept pipeline cannot work without and
-// that are nil. A production wiring must call it and refuse to run with any
-// missing (a nil Headers, for example, rejects every bundle at the first
-// sample): optional pieces are only Config.AllowedExtensions and
-// Config.WillBeChecked (nil means "no restriction" and "every path").
+// Missing lists the required collaborators that are nil. Without one the
+// accept pipeline cannot work (a nil State or Library panics at first use, a
+// nil Headers rejects every bundle at its first sample). The optional ones are
+// listed on Config.
func (c Config) Missing() []string {
var out []string
add := func(nilDep bool, name string) {
@@ -51,12 +54,7 @@ func (c Config) Missing() []string {
add(c.Stager == nil, "Stager")
add(c.Prober == nil, "Prober")
add(c.State == nil, "State")
- add(c.Index == nil, "Index")
- add(c.Fetcher == nil, "Fetcher")
add(c.Headers == nil, "Headers")
- add(c.Recorder == nil, "Recorder")
- add(c.Announcer == nil, "Announcer")
- add(c.Reporter == nil, "Reporter")
add(c.HealthEnabled == nil, "HealthEnabled")
return out
}
@@ -100,6 +98,16 @@ type targetLock struct {
refs int
}
+// NewChecked is New that refuses a Config with a required collaborator
+// missing: an importer built without one would panic or reject everything at
+// first use.
+func NewChecked(cfg Config) (*Importer, error) {
+ if missing := cfg.Missing(); len(missing) > 0 {
+ return nil, fmt.Errorf("bundleimport: required collaborators missing: %s", strings.Join(missing, ", "))
+ }
+ return New(cfg), nil
+}
+
// New builds an Importer. A zero Policy means DefaultPolicy.
func New(cfg Config) *Importer {
if cfg.Policy == (Policy{}) {
diff --git a/internal/peernet/bundleimport/newchecked_test.go b/internal/peernet/bundleimport/newchecked_test.go
new file mode 100644
index 000000000..45514c779
--- /dev/null
+++ b/internal/peernet/bundleimport/newchecked_test.go
@@ -0,0 +1,24 @@
+package bundleimport
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// An importer is never built without a collaborator it cannot work without:
+// a nil State or Library would panic at first use.
+func TestNewCheckedRefusesAMissingRequiredCollaborator(t *testing.T) {
+ imp, err := NewChecked(Config{})
+ require.Error(t, err)
+ assert.Nil(t, imp)
+ for _, name := range []string{"Library", "Stager", "Prober", "State", "Headers", "HealthEnabled"} {
+ assert.Contains(t, err.Error(), name)
+ }
+ e := newEnv(t)
+ cfg := e.imp.cfg
+ cfg.Prober = e.prober
+ _, err = NewChecked(cfg)
+ require.NoError(t, err, "optional collaborators (Index, Fetcher, Recorder) may be nil")
+}
diff --git a/internal/peernet/bundleimport/parts.go b/internal/peernet/bundleimport/parts.go
index 1402b5218..f394b614d 100644
--- a/internal/peernet/bundleimport/parts.go
+++ b/internal/peernet/bundleimport/parts.go
@@ -215,7 +215,11 @@ func (i *Importer) verifyParts(ctx context.Context, b *bundle.Bundle, tables []s
for n, r := range picks {
f := &b.Files[r.file]
ps := partsOfSliced(f)[r.part]
- w := &partWork{ref: r, segs: ps, raw: rawView(ps), name: fmt.Sprintf("part-%d.bin", n), upTo: ps[0].Start, offset: ps[0].Start, table: byTable[[2]int{r.file, r.part}]}
+ tbl, ok := byTable[[2]int{r.file, r.part}]
+ if !ok {
+ return ReasonInvalid, false, fmt.Errorf("part %d of file %d has no sample table", r.part, r.file)
+ }
+ w := &partWork{ref: r, segs: ps, raw: rawView(ps), name: fmt.Sprintf("part-%d.bin", n), upTo: ps[0].Start, offset: ps[0].Start, table: tbl}
if r.kind == partRaw {
w.upTo = 0
} else {
diff --git a/internal/peernet/bundleimport/pathobserver.go b/internal/peernet/bundleimport/pathobserver.go
index 39b448723..072127d28 100644
--- a/internal/peernet/bundleimport/pathobserver.go
+++ b/internal/peernet/bundleimport/pathobserver.go
@@ -11,7 +11,7 @@ import (
// renameGrace is how long a deleted file's row waits for the matching write
// (the metadata service reports a rename as a delete followed by a write).
-const renameGrace = 3 * time.Second
+var renameGrace = 3 * time.Second
// PathObserver keeps the peer-bundle state consistent with the library when
// the user deletes, moves or renames a file or folder: a deleted file loses its
diff --git a/internal/peernet/bundleimport/pathobserver_test.go b/internal/peernet/bundleimport/pathobserver_test.go
index 93b7e9621..9603fa834 100644
--- a/internal/peernet/bundleimport/pathobserver_test.go
+++ b/internal/peernet/bundleimport/pathobserver_test.go
@@ -3,6 +3,7 @@ package bundleimport
import (
"context"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -67,3 +68,32 @@ func TestPathObserver_RenamedFolderMovesRows(t *testing.T) {
obs.DirectoryDeleted(ctx, "movies/Other")
assert.Zero(t, e.stateRows())
}
+
+// The library reports deletes under whatever spelling the caller used (the
+// HTTP API passes a leading slash); the row is keyed by the clean path.
+func TestPathObserver_DeleteWithLeadingSlashRemovesTheRow(t *testing.T) {
+ e, res := newObserved(t)
+ ctx := context.Background()
+ obs := e.imp.PathObserver()
+ require.NoError(t, e.lib.DeleteFileMetadata(ctx, res.Paths[0]))
+ obs.FileDeleted(ctx, "/"+res.Paths[0])
+ obs.Flush()
+ _, ok, _ := e.repo.Get(ctx, res.Paths[0])
+ assert.False(t, ok, "the deleted file's row is gone")
+}
+
+// Through the real metadata service: a user deleting the file ends its state
+// row once the rename grace has passed, with no further attempt needed.
+func TestPathObserver_UserDeleteThroughTheLibraryEndsTheStateRow(t *testing.T) {
+ old := renameGrace
+ renameGrace = 50 * time.Millisecond
+ t.Cleanup(func() { renameGrace = old })
+ e, res := newObserved(t)
+ ctx := context.Background()
+ e.lib.SetSecondaryFileObserver(e.imp.PathObserver())
+ require.NoError(t, e.lib.DeleteFileMetadata(ctx, res.Paths[0]))
+ require.Eventually(t, func() bool {
+ _, ok, _ := e.repo.Get(ctx, res.Paths[0])
+ return !ok
+ }, 2*time.Second, 20*time.Millisecond, "the row of a deleted file must not outlive it")
+}
diff --git a/internal/peernet/bundleimport/refund_test.go b/internal/peernet/bundleimport/refund_test.go
new file mode 100644
index 000000000..d28c50698
--- /dev/null
+++ b/internal/peernet/bundleimport/refund_test.go
@@ -0,0 +1,43 @@
+package bundleimport
+
+import (
+ "context"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// A bundle the local file policy refuses before any article is read gives the
+// publisher's fetch budget back; one that failed on article evidence does not.
+func TestAcceptRefundsTheFetchBudgetOnlyForLocalPolicyRejections(t *testing.T) {
+ e := newEnv(t)
+ b := e.mkBundle("refund", fileSpec{name: "data.bin", plain: junkPlain(512), segSize: 128})
+ _, err := e.imp.Accept(context.Background(), e.req(b))
+ require.Error(t, err)
+ assert.Equal(t, ReasonNoMediaFile, ReasonOf(err))
+ assert.Len(t, e.rep.refunds, 1, "no article was read: the attempt is refunded")
+
+ e2 := newEnv(t)
+ b2 := e2.mkBundle("fail", fileSpec{name: "movie.mkv", plain: junkPlain(512), segSize: 128})
+ _, err = e2.imp.Accept(context.Background(), e2.req(b2))
+ require.Error(t, err)
+ assert.NotEqual(t, ReasonNoMediaFile, ReasonOf(err))
+ assert.Empty(t, e2.rep.refunds, "an attempt that read articles keeps its charge")
+}
+
+// The import counters say how many parts and extents of accepted bundles were
+// bound to an archive header and how many could not be.
+func TestCountersExposeHeaderBinding(t *testing.T) {
+ e := newEnv(t)
+ payload := mkvPlain(1500)
+ s := e.mkStoredSet(setSpec{tag: "hb", kind: "rar5", payload: payload, cuts: evenCuts(len(payload), 3), scheme: 1, extraVol: -1})
+ res, err := e.imp.Accept(context.Background(), e.req(s.b))
+ require.NoError(t, err)
+ require.Equal(t, 3, res.Cost.PartsBound)
+ c := e.imp.Counters()
+ assert.Equal(t, uint64(3), c.PartsHeaderBound)
+ assert.Equal(t, uint64(res.Cost.PartsUnbound), c.PartsHeaderUnbound)
+ assert.Equal(t, uint64(res.Cost.ExtentsBound), c.ExtentsHeaderBound)
+ assert.Equal(t, uint64(res.Cost.ExtentsUnbound), c.ExtentsHeaderUnbound)
+}
diff --git a/internal/peernet/bundleimport/types.go b/internal/peernet/bundleimport/types.go
index fcd18b627..4d655b066 100644
--- a/internal/peernet/bundleimport/types.go
+++ b/internal/peernet/bundleimport/types.go
@@ -445,17 +445,24 @@ type Counters struct {
FullCheckFailed uint64
Announced uint64
BadBundleReports uint64
+ // PartsHeaderBound and PartsHeaderUnbound (ExtentsHeaderBound and
+ // ExtentsHeaderUnbound for nested-layout extents) add up, over the
+ // accepted bundles, the parts bound to an archive header and those that
+ // could not be.
+ PartsHeaderBound, PartsHeaderUnbound uint64
+ ExtentsHeaderBound, ExtentsHeaderUnbound uint64
}
type counters struct {
- mu sync.Mutex
- accepted uint64
- already uint64
- rejected map[Reason]uint64
- passed uint64
- failed uint64
- announce uint64
- reports uint64
+ partsBound, partsUnbound, extBound, extUnbound uint64
+ mu sync.Mutex
+ accepted uint64
+ already uint64
+ rejected map[Reason]uint64
+ passed uint64
+ failed uint64
+ announce uint64
+ reports uint64
}
func (c *counters) add(f func(*counters)) {
@@ -472,7 +479,8 @@ func (c *counters) snapshot() Counters {
rej[k] = v
}
return Counters{Accepted: c.accepted, AlreadyImported: c.already, Rejected: rej, FullCheckPassed: c.passed,
- FullCheckFailed: c.failed, Announced: c.announce, BadBundleReports: c.reports}
+ FullCheckFailed: c.failed, Announced: c.announce, BadBundleReports: c.reports,
+ PartsHeaderBound: c.partsBound, PartsHeaderUnbound: c.partsUnbound, ExtentsHeaderBound: c.extBound, ExtentsHeaderUnbound: c.extUnbound}
}
// YencSamples returns the sampling bound with its default.
@@ -499,3 +507,22 @@ func (p Policy) YencTimeoutOrDefault() time.Duration {
}
return p.YencTimeout
}
+
+// FetchRefunder is an optional capability of the Reporter: it gives the
+// per-publisher fetch budget back for a bundle the local policy refused
+// before any article was read (see localPolicyReason).
+type FetchRefunder interface {
+ RefundFetch(ctx context.Context, o Origin)
+}
+
+// localPolicyReason says a rejection came from the local rules alone: it cost
+// the network nothing beyond the bundle itself and says nothing about the
+// publisher, so it must not use up the publisher's fetch budget.
+func localPolicyReason(r Reason) bool {
+ switch r {
+ case ReasonTooLarge, ReasonTooManyFiles, ReasonForbiddenFile, ReasonSidecarTooLarge, ReasonNoMediaFile,
+ ReasonHealthDisabled, ReasonBadTarget, ReasonCollision, ReasonNZBMismatch:
+ return true
+ }
+ return false
+}
diff --git a/internal/peernet/bundlex/build_bench_test.go b/internal/peernet/bundlex/build_bench_test.go
index 825dd6d72..c57bf5f95 100644
--- a/internal/peernet/bundlex/build_bench_test.go
+++ b/internal/peernet/bundlex/build_bench_test.go
@@ -17,8 +17,8 @@ func BenchmarkBuild75kSegments(b *testing.B) {
for i := range video {
video[i] = fmt.Sprintf("part%d-%05dof75000.%x@news.example.net", i, i, uint64(i)*2654435761)
}
- lib := &fakeLib{scheme: database.PeernetSchemeNZB, paths: []string{"p"},
- files: map[string]*metapb.FileMetadata{"p": plainFile("big.mkv", video)}, nzb: [][]string{video}}
+ lib := &fakeLib{scheme: database.PeernetSchemeNZB, paths: []string{"p/big.mkv"},
+ files: map[string]*metapb.FileMetadata{"p/big.mkv": plainFile("big.mkv", video)}, nzb: [][]string{video}}
rel := releaseHexOfNZB(lib.nzb)
bl := NewBuilder(BuilderConfig{Source: lib, Library: lib, Patches: NoPatches{}})
b.ResetTimer()
diff --git a/internal/peernet/bundlex/builder.go b/internal/peernet/bundlex/builder.go
index 839e0dab8..79f2a5745 100644
--- a/internal/peernet/bundlex/builder.go
+++ b/internal/peernet/bundlex/builder.go
@@ -5,6 +5,8 @@ import (
"encoding/hex"
"errors"
"fmt"
+ "path"
+ "strings"
"sync"
"time"
@@ -13,6 +15,7 @@ import (
metapb "github.com/kipsilabs/altmount/internal/metadata/proto"
"github.com/kipsilabs/altmount/internal/nzbname"
"github.com/kipsilabs/altmount/internal/peernet/bundle"
+ "github.com/kipsilabs/altmount/internal/peernet/bundleimport"
"github.com/kipsilabs/altmount/internal/peernet/contentid"
"github.com/kipsilabs/altmount/internal/peernet/records"
)
@@ -30,6 +33,7 @@ const (
ReasonNoPostDate = "no_post_date"
ReasonNoFiles = "no_files"
ReasonSelfCheckFailed = "self_check_failed"
+ ReasonNoMediaFile = "no_media_file"
maxBuiltCacheBytes = 32 << 20
builtTTL = 10 * time.Minute
defaultBuildWait = 30 * time.Second
@@ -415,7 +419,7 @@ func (b *Builder) build(ctx context.Context, releaseHex, title string, posted ti
if len(m.KnownHoles) > 0 {
return nil, notShareable(ReasonKnownHoles)
}
- fi := bundle.FileInput{Meta: m, PatchedLocally: b.anyPatched(m)}
+ fi := bundle.FileInput{Meta: m, Name: path.Base(strings.ReplaceAll(p, "\\", "/")), PatchedLocally: b.anyPatched(m)}
pw, fromNZB := b.cfg.Source.NZBPassword(p, m)
fi.PasswordFromNZB = fromNZB
if fromNZB && in.NZBPassword == "" {
@@ -447,6 +451,10 @@ func (b *Builder) build(ctx context.Context, releaseHex, title string, posted ti
if bd.ReleaseID != rid {
return nil, notShareable(ReasonReleaseMismatch)
}
+ if _, reason, perr := bundleimport.SelectFiles(bd, nil, bundleimport.Policy{}); perr != nil {
+ // A receiver applies this same policy before reading any article.
+ return nil, notShareable(string(reason))
+ }
wire, id, err := bundle.Pack(bd)
if err != nil {
if errors.Is(err, bundle.ErrTooLarge) {
diff --git a/internal/peernet/bundlex/builder_test.go b/internal/peernet/bundlex/builder_test.go
index a881ca89c..3cab6aca1 100644
--- a/internal/peernet/bundlex/builder_test.go
+++ b/internal/peernet/bundlex/builder_test.go
@@ -67,7 +67,7 @@ func TestBuildFilesSchemeNeverFallsBackFromNZB(t *testing.T) {
f := plainFile("a.mkv", ids("a", 5))
fid, _ := contentid.FileID(f)
frid, _ := contentid.ReleaseIDFromFiles([]contentid.ID{fid})
- lib2 := &fakeLib{scheme: database.PeernetSchemeFiles, paths: []string{"p"}, files: map[string]*metapb.FileMetadata{"p": f}}
+ lib2 := &fakeLib{scheme: database.PeernetSchemeFiles, paths: []string{"p/a.mkv"}, files: map[string]*metapb.FileMetadata{"p/a.mkv": f}}
got, err := newTestBuilder(t, lib2, NoPatches{}).Build(context.Background(), frid.String(), "T", posted)
if err != nil || got.Scheme != bundle.SchemeFiles {
t.Fatalf("files scheme: %v %v", got, err)
@@ -292,3 +292,35 @@ func TestBuildUnrecoverablePartLayoutIsCountedNotShareable(t *testing.T) {
t.Fatalf("not counted: %v", b.Status().NotShareable)
}
}
+
+// An archive-stored file has no OriginalFilename in its metadata; the bundle
+// must carry the file's virtual file name instead.
+func TestBuildNamesFilesByVirtualPath(t *testing.T) {
+ vid := ids("vid", 20)
+ lib := &fakeLib{scheme: database.PeernetSchemeNZB, paths: []string{"movies/R/Movie.2020.mkv"},
+ files: map[string]*metapb.FileMetadata{"movies/R/Movie.2020.mkv": plainFile("", vid)}, nzb: [][]string{vid}}
+ rel := releaseHexOfNZB(lib.nzb)
+ got, err := newTestBuilder(t, lib, NoPatches{}).Build(context.Background(), rel, "T", posted)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rid, _ := contentid.ParseID(rel)
+ bd, err := bundle.Verify(got.Wire, got.ID, rid)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if bd.Files[0].Name != "Movie.2020.mkv" {
+ t.Fatalf("file name = %q, want the virtual file name", bd.Files[0].Name)
+ }
+}
+
+// A release the receiver's file policy would certainly refuse is not announced.
+func TestBuildRefusesReleaseWithoutMediaFile(t *testing.T) {
+ d := ids("d", 20)
+ lib := &fakeLib{scheme: database.PeernetSchemeNZB, paths: []string{"movies/R/payload.bin"},
+ files: map[string]*metapb.FileMetadata{"movies/R/payload.bin": plainFile("", d)}, nzb: [][]string{d}}
+ _, err := newTestBuilder(t, lib, NoPatches{}).Build(context.Background(), releaseHexOfNZB(lib.nzb), "T", posted)
+ if r := notShareableReason(t, err); r != ReasonNoMediaFile {
+ t.Fatalf("reason = %s", r)
+ }
+}
diff --git a/internal/peernet/bundlex/fetch.go b/internal/peernet/bundlex/fetch.go
index 782991415..a565bb92f 100644
--- a/internal/peernet/bundlex/fetch.go
+++ b/internal/peernet/bundlex/fetch.go
@@ -690,3 +690,11 @@ func (f *Fetcher) ReportVerifiedBundle(nw netid.NetworkID, p identity.NodeID) {
defer f.mu.Unlock()
delete(f.fcStrikes, strikeKey(nw, p))
}
+
+// RefundFetch gives one attempt back to the per-publisher budget of p. The
+// caller uses it when the bundle was refused by the local file policy alone,
+// before any article was read: such an attempt must not eat into the budget of
+// later, legitimate plays. The node-wide budget is not refunded.
+func (f *Fetcher) RefundFetch(nw netid.NetworkID, p identity.NodeID) {
+ f.pub.give(strikeKey(nw, p), 1)
+}
diff --git a/internal/peernet/bundlex/fetch_test.go b/internal/peernet/bundlex/fetch_test.go
index 649dd1ad2..a86dff92c 100644
--- a/internal/peernet/bundlex/fetch_test.go
+++ b/internal/peernet/bundlex/fetch_test.go
@@ -133,8 +133,8 @@ func TestFetchVerifiesAndReportsSource(t *testing.T) {
func TestFetchRejectsSubstitutedGarbageOversizedAndStrikes(t *testing.T) {
// A valid bundle of another release, served under the announced IDs.
- lib2 := &fakeLib{scheme: database.PeernetSchemeNZB, paths: []string{"x"}, nzb: [][]string{ids("z", 4)}}
- lib2.files = map[string]*metapb.FileMetadata{"x": plainFile("z.mkv", ids("z", 4))}
+ lib2 := &fakeLib{scheme: database.PeernetSchemeNZB, paths: []string{"x/z.mkv"}, nzb: [][]string{ids("z", 4)}}
+ lib2.files = map[string]*metapb.FileMetadata{"x/z.mkv": plainFile("z.mkv", ids("z", 4))}
other, err := newTestBuilder(t, lib2, NoPatches{}).Build(context.Background(), releaseHexOfNZB(lib2.nzb), "T", posted)
if err != nil {
t.Fatal(err)
@@ -602,3 +602,27 @@ func TestFetchReportsPublisherLimitAndBackOffAsTheirOwnErrors(t *testing.T) {
t.Fatalf("a backed-off publisher is ErrBackedOff (still an ErrNoOffers): %v", err)
}
}
+
+// A bundle the local policy refused before any article was read does not use
+// up the publisher's budget.
+func TestRefundFetchGivesThePublisherBudgetBack(t *testing.T) {
+ e := newFetchEnv(t, FetchLimits{PerPublisherPerHour: 1, PerPublisherBurst: 1})
+ addr, id := honest(t, e)
+ e.publisher(addr, id, time.Hour)
+ if _, _, err := e.f.Fetch(context.Background(), e.rel); err != nil {
+ t.Fatal(err)
+ }
+ e.f.RefundFetch(nwA, id)
+ if _, _, err := e.f.Fetch(context.Background(), e.rel); err != nil {
+ t.Fatalf("a refunded attempt must leave the budget intact: %v", err)
+ }
+}
+
+// The defaults must let a user browse and play: a burst of ten and a minute
+// per attempt sustained, per publisher.
+func TestDefaultFetchLimitsSuitBrowsing(t *testing.T) {
+ d := DefaultFetchLimits()
+ if d.PerPublisherBurst < 10 || d.PerPublisherPerHour < 60 {
+ t.Fatalf("per publisher: burst %d, %d/hour", d.PerPublisherBurst, d.PerPublisherPerHour)
+ }
+}
diff --git a/internal/peernet/bundlex/limits.go b/internal/peernet/bundlex/limits.go
index e126e4130..75d99c6f7 100644
--- a/internal/peernet/bundlex/limits.go
+++ b/internal/peernet/bundlex/limits.go
@@ -101,7 +101,7 @@ type FetchLimits struct {
// DefaultFetchLimits returns the defaults.
func DefaultFetchLimits() FetchLimits {
return FetchLimits{
- MaxAttempts: 3, GlobalPerHour: 60, GlobalBurst: 10, PerPublisherPerHour: 12, PerPublisherBurst: 3,
+ MaxAttempts: 3, GlobalPerHour: 60, GlobalBurst: 10, PerPublisherPerHour: 60, PerPublisherBurst: 10,
MaxConcurrent: 2, Timeout: 60 * time.Second,
BackoffBase: 10 * time.Minute, BackoffMax: 2 * time.Hour, SoftBackoff: 5 * time.Minute,
FullCheckBackoffBase: 30 * time.Minute, FullCheckBackoffMax: 6 * time.Hour, FullCheckDecay: 24 * time.Hour,
diff --git a/internal/peernet/service/bundles.go b/internal/peernet/service/bundles.go
index eeae1f659..78e38d6e7 100644
--- a/internal/peernet/service/bundles.go
+++ b/internal/peernet/service/bundles.go
@@ -82,6 +82,11 @@ type BundleImportStatus struct {
Rejected map[string]int64
FullCheckPassed, FullCheckFailed int64
Announced, BadBundleReports int64
+ // PartsHeaderBound and PartsHeaderUnbound (and the extent pair) say how
+ // many parts and nested-layout extents of accepted bundles were bound to
+ // an archive header at accept time and how many could not be.
+ PartsHeaderBound, PartsHeaderUnbound int64
+ ExtentsHeaderBound, ExtentsHeaderUnbound int64
// Pending files still owe their full check; Stale are pending past the
// maximum age (they are failed, never announced).
Pending, Stale int64
@@ -314,6 +319,25 @@ func (s *Service) FetchBundleWith(ctx context.Context, releaseID string, opts bu
return n.fetcher.FetchWith(ctx, releaseID, opts)
}
+// RefundBundleFetch gives one attempt back to the per-publisher fetch budget
+// (a bundle refused by the local file policy before any article was read).
+// Unparseable identifiers are ignored.
+func (s *Service) RefundBundleFetch(networkID, publisherID string) {
+ n := s.currentNode()
+ if n == nil || n.fetcher == nil {
+ return
+ }
+ nw, err := netid.ParseNetworkID(networkID)
+ if err != nil {
+ return
+ }
+ pub, err := identity.ParseNodeID(publisherID)
+ if err != nil {
+ return
+ }
+ n.fetcher.RefundFetch(nw, pub)
+}
+
// NoteBundleFullCheck tells the fetcher how the full local check of a release
// imported from publisherID's bundle in networkID ended. A failure backs that
// publisher off for bundle fetches only (never record sync, never a purge); a
diff --git a/internal/peernet/service/catchup_test.go b/internal/peernet/service/catchup_test.go
index a1958fb6d..6427f8e39 100644
--- a/internal/peernet/service/catchup_test.go
+++ b/internal/peernet/service/catchup_test.go
@@ -29,6 +29,9 @@ func TestCatchUpPublishesHealthyIndexedReleasesNeverPublished(t *testing.T) {
if _, ok, _ := f.state.Get(ctx, nw, database.PeernetRecordKindRelease, pubRelease, ""); !ok {
t.Fatal("the healthy indexed release must be published by the catch-up pass")
}
+ if f.status(t).Publish.Published["release"] == 0 {
+ t.Fatalf("a catch-up publish must be counted as published: %+v", f.status(t).Publish.Published)
+ }
if _, ok, _ := f.state.Get(ctx, nw, database.PeernetRecordKindAttestation, pubRelease, pubFile); ok {
t.Fatal("an attestation is never published without a new check")
}
diff --git a/internal/peernet/service/publishing.go b/internal/peernet/service/publishing.go
index 3225390ad..b7f355d66 100644
--- a/internal/peernet/service/publishing.go
+++ b/internal/peernet/service/publishing.go
@@ -567,6 +567,8 @@ func (p *pipeline) reannounceOne(ctx context.Context, t publish.Tracked, relHex
p.stats.mu.Lock()
p.stats.reannounced++
p.stats.mu.Unlock()
+ // A re-announce and a catch-up publish are records sent: count them as such.
+ p.stats.add(p.stats.published, kindName(records.KindRelease))
return true
}
diff --git a/internal/testsupport/mediagen/mediagen.go b/internal/testsupport/mediagen/mediagen.go
new file mode 100644
index 000000000..a612cb33e
--- /dev/null
+++ b/internal/testsupport/mediagen/mediagen.go
@@ -0,0 +1,56 @@
+// Package mediagen builds small, structurally valid media files for tests.
+package mediagen
+
+// MKV returns a structurally valid Matroska file of exactly size bytes
+// (size >= 200): EBML header, a Segment ending at the file end, a SeekHead
+// pointing at Info and at Cues (the last element), and a Cluster of
+// SimpleBlocks between them.
+func MKV(size int) []byte {
+ u32 := func(v int) []byte { return []byte{byte(v >> 24), byte(v >> 16), byte(v >> 8), byte(v)} }
+ seek := func(id []byte, pos int) []byte {
+ body := append(append([]byte{0x53, 0xAB, 0x84}, id...), append([]byte{0x53, 0xAC, 0x84}, u32(pos)...)...)
+ return append([]byte{0x4D, 0xBB, byte(0x80 | len(body))}, body...)
+ }
+ info := []byte{0x15, 0x49, 0xA9, 0x66}
+ cues := []byte{0x1C, 0x53, 0xBB, 0x6B}
+ ebml := append([]byte{0x1A, 0x45, 0xDF, 0xA3, 0x8B, 0x42, 0x82, 0x88}, "matroska"...)
+ segStart := len(ebml) + 12
+ seekHead := append([]byte{0x11, 0x4D, 0x9B, 0x74, byte(0x80 | 34)}, append(seek(info, 39), seek(cues, size-segStart-9)...)...)
+ out := append([]byte{}, ebml...)
+ segSize := size - segStart
+ out = append(out, 0x18, 0x53, 0x80, 0x67, 0x01, 0, byte(segSize>>40), byte(segSize>>32), byte(segSize>>24), byte(segSize>>16), byte(segSize>>8), byte(segSize))
+ out = append(out, seekHead...)
+ out = append(out, append(append([]byte{}, info...), 0x84, 0, 0, 0, 0)...)
+ fill := size - len(out) - 12 - 9
+ out = append(out, 0x1F, 0x43, 0xB6, 0x75, 0x01, 0, byte(fill>>40), byte(fill>>32), byte(fill>>24), byte(fill>>16), byte(fill>>8), byte(fill))
+ out = append(out, SimpleBlocks(fill)...)
+ return append(out, append(append([]byte{}, cues...), 0x84, 0, 0, 0, 0)...)
+}
+
+// AVI returns a RIFF/AVI container of size bytes: the container signature and
+// header list followed by filler.
+func AVI(size int) []byte {
+ out := append([]byte("RIFF\x00\x00\x00\x00AVI LIST\x00\x00\x00\x00hdrlavih"), SimpleBlocks(size-28)...)
+ n := uint32(len(out) - 8)
+ out[4], out[5], out[6], out[7] = byte(n), byte(n>>8), byte(n>>16), byte(n>>24)
+ return out
+}
+
+// SimpleBlocks returns exactly n bytes (n >= 7) of consecutive Matroska
+// SimpleBlock elements, what the body of a real Cluster looks like.
+func SimpleBlocks(n int) []byte {
+ out := make([]byte, 0, n)
+ for k := 0; len(out) < n; k++ {
+ remaining := n - len(out)
+ want := min(remaining, 90+(k*37)%300)
+ if remaining-want < 7 {
+ want = remaining
+ }
+ body := want - 3
+ out = append(out, 0xA3, 0x40|byte(body>>8), byte(body), 0x81, byte(k>>8), byte(k), 0x80)
+ for j := 0; j < body-4; j++ {
+ out = append(out, byte(1+(len(out)*7+k)%250))
+ }
+ }
+ return out
+}