amadan
objstore: the scan result object, version-bound and guarded
Somewhere to put a verdict the object itself cannot carry. tag() declines to write when an object already holds eight foreign tags — S3 caps an object at ten — or when the pre-write tag read is denied. Both refusals are deliberate and both are fine for a sweep that re-lists every 24h. A queue item is not: the drain scans, gets a real verdict and deletes the item, so without this the status door would answer pending forever while the callback said clean, and the app's declared source of truth would never arrive. Keyed by sha256 of the object key, because the door asks by key and ULID-named queue items would mean listing the whole prefix on every poll. And version-bound, which is the part this design has now got wrong three times in review. Without it: v1 scanned, tag not written because the cap fired, result says clean; the tenant overwrites with v2; the tag is still absent, the door falls through, and it answers clean for bytes nobody has read — the laundering the verdict tag's own binding exists to stop, reintroduced by the mechanism meant to close a gap in it. The comparison lives in ReadScanResult so no door and no app handles the token, and nobody can forget to make it. Mutation-checked. Read-then-CAS, not create-only. Put has exactly two modes and the second write for a key is a replace — v1 unrecordable, tenant uploads v2, v2 also unrecordable — so a create-only write there conflicts and silently drops the newer verdict. Under the app's prefix, so PurgeApp removes it and the claim gate sees it. A stranded one is worse than a stranded queue item: that gets drained next tick because its index entry survives, while this is found by nothing, blocks the app name, and — because materialize counts any key under an app prefix and fails toward keeping the row — resurrects the deleted app on the next boot. The caller must re-read store.json immediately before writing, and the doc comment says so.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2 files changed,
+294
−0
internal/objstore/scanresult.go+154 −0internal/objstore/scanresult_test.go+140 −0
diff --git a/internal/objstore/scanresult.go b/internal/objstore/scanresult.go| new file mode 100644 |
| index 000000000..95339b564 |
| --- /dev/null |
| +++ b/internal/objstore/scanresult.go |
| @@ -0,0 +1,154 @@ |
| +// The scan result object: apps/<acct>/<app>/scan-results/<sha256(key)>.json. |
| +// |
| +// It exists for one narrow case, and the case is not hypothetical. tag() |
| +// declines to write a verdict when the object already carries eight foreign |
| +// tags (S3 caps an object at ten) or when the pre-write tag read is denied. |
| +// Both refusals are deliberate — they stop one object stalling a whole sweep — |
| +// and the sweep tolerates them because it re-lists every 24h. |
| +// |
| +// A queue item does not. The on-demand drain scans, gets a real verdict, and |
| +// deletes the item; without somewhere else to put that verdict the status door |
| +// would answer "pending" forever while the callback said "clean". The app's |
| +// declared source of truth would never arrive, and the two answers it did get |
| +// would disagree. |
| +// |
| +// KEYED BY sha256(key), NOT by a queue id: the door asks by key, and ULID-named |
| +// queue items would mean listing the whole prefix on every poll. |
| +// |
| +// AND VERSION-BOUND, which is the part that has been got wrong three times in |
| +// this design's reviews. Without it: v1 scanned, tag not written because the |
| +// cap fired, result says clean; the tenant overwrites with v2; the tag is still |
| +// absent, the door falls through to the result object, and answers clean for |
| +// bytes nobody has read. That is precisely the laundering the verdict tag's own |
| +// version binding exists to stop, reintroduced by the mechanism meant to close |
| +// a gap in it. |
| +package objstore |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// ScanResultSchema is pinned; an unknown schema refuses rather than guessing. |
| +const ScanResultSchema = 1 |
| + |
| +// ScanResult is a verdict that was reached but could not be written onto the |
| +// object it describes. |
| +type ScanResult struct { |
| + Schema int `json:"schema"` |
| + Key string `json:"key"` |
| + Verdict string `json:"verdict"` |
| + // Version is the object version this verdict judged. A result whose |
| + // version does not match the object's current one is not an answer about |
| + // the object that is there now. |
| + Version string `json:"version"` |
| + Engine string `json:"engine,omitempty"` |
| + Signature string `json:"signature,omitempty"` |
| + At time.Time `json:"at"` |
| +} |
| + |
| +// ScanResultKey is where one key's unrecorded verdict lives. |
| +// |
| +// Under the APP's prefix, so PurgeApp removes it with everything else and the |
| +// name-claim gate can see it. A top-level results prefix would survive app |
| +// deletion invisibly, which is the same argument that keeps the queue here. |
| +func ScanResultKey(acct, app, objectKey string) string { |
| + sum := sha256.Sum256([]byte(objectKey)) |
| + return release.AppPrefix(acct, app) + "scan-results/" + hex.EncodeToString(sum[:]) + ".json" |
| +} |
| + |
| +// ScanResultsPrefix is the whole set, for the age sweep and for deletion when |
| +// an app stops being scannable. |
| +func ScanResultsPrefix(acct, app string) string { |
| + return release.AppPrefix(acct, app) + "scan-results/" |
| +} |
| + |
| +// WriteScanResult records a verdict the object itself could not carry. |
| +// |
| +// READ-THEN-CAS, not create-only. release.Store.Put has exactly two modes, and |
| +// the second write for a key is a REPLACE: v1 unrecordable, tenant uploads v2, |
| +// v2 also unrecordable. A create-only write there returns ErrCASConflict and |
| +// silently drops the v2 verdict, leaving the door on "pending" until a sweep |
| +// happens to tag it. |
| +// |
| +// The caller must have re-read store.json immediately before calling this. The |
| +// race is the one writeSidecar documents — read the record, scan for seconds, |
| +// a teardown and PurgeApp land, and a write under the purged prefix succeeds — |
| +// but the consequence here is worse than a stranded sidecar. A stranded queue |
| +// item is drained on the next tick because its index entry survives; a stranded |
| +// result object is found by nothing, blocks the app-name claim gate, and |
| +// because materialize counts any key under an app prefix and fails toward |
| +// keeping the row, it RESURRECTS the deleted app on the next console boot. |
| +func WriteScanResult(ctx context.Context, s release.Store, acct, app string, res *ScanResult) error { |
| + res.Schema = ScanResultSchema |
| + body, err := json.MarshalIndent(res, "", " ") |
| + if err != nil { |
| + return err |
| + } |
| + body = append(body, '\n') |
| + key := ScanResultKey(acct, app, res.Key) |
| + |
| + for attempt := 0; attempt < 2; attempt++ { |
| + _, etag, err := s.Get(ctx, key) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return fmt.Errorf("objstore: read scan result %s/%s: %w", acct, app, err) |
| + } |
| + if errors.Is(err, os.ErrNotExist) { |
| + etag = "" // create-only |
| + } |
| + if _, err := s.Put(ctx, key, body, etag); err != nil { |
| + if errors.Is(err, release.ErrCASConflict) { |
| + continue // somebody moved under us; re-read and try once more |
| + } |
| + return fmt.Errorf("objstore: write scan result %s/%s: %w", acct, app, err) |
| + } |
| + return nil |
| + } |
| + return fmt.Errorf("objstore: write scan result %s/%s: lost the race twice", acct, app) |
| +} |
| + |
| +// ReadScanResult answers only about currentVersion. A result recorded for |
| +// different bytes is ABSENT, not stale-but-usable: the comparison lives here so |
| +// no door and no app ever handles the version token, and so nobody can forget |
| +// to make it. |
| +func ReadScanResult(ctx context.Context, s release.Store, acct, app, objectKey, currentVersion string) (*ScanResult, error) { |
| + raw, _, err := s.Get(ctx, ScanResultKey(acct, app, objectKey)) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, nil |
| + } |
| + return nil, err |
| + } |
| + var res ScanResult |
| + if err := json.Unmarshal(raw, &res); err != nil { |
| + return nil, fmt.Errorf("objstore: scan result %s/%s: %w", acct, app, err) |
| + } |
| + if res.Schema != ScanResultSchema { |
| + return nil, fmt.Errorf("objstore: scan result %s/%s: schema %d is not %d", |
| + acct, app, res.Schema, ScanResultSchema) |
| + } |
| + if res.Version != currentVersion { |
| + // Recorded about bytes that are no longer there. |
| + return nil, nil |
| + } |
| + return &res, nil |
| +} |
| + |
| +// DeleteScanResult drops one key's result — called when a later scan DOES |
| +// persist a verdict onto the object, so the fallback cannot outlive the real |
| +// answer. |
| +func DeleteScanResult(ctx context.Context, s release.Store, acct, app, objectKey string) error { |
| + err := s.Delete(ctx, ScanResultKey(acct, app, objectKey)) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return err |
| + } |
| + return nil |
| +} |
diff --git a/internal/objstore/scanresult_test.go b/internal/objstore/scanresult_test.go| new file mode 100644 |
| index 000000000..7993429f7 |
| --- /dev/null |
| +++ b/internal/objstore/scanresult_test.go |
| @@ -0,0 +1,140 @@ |
| +package objstore |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +func resultFixture(key, verdict, version string) *ScanResult { |
| + return &ScanResult{ |
| + Key: key, Verdict: verdict, Version: version, |
| + Engine: "clamav", Signature: "ClamAV 1.4.1/27500/x", At: time.Now().UTC(), |
| + } |
| +} |
| + |
| +// THE test for this file, and the third appearance of this bug class in the |
| +// design's reviews. |
| +// |
| +// Without the version binding: v1 scanned, the tag not written because the |
| +// ten-tag cap fired, the result says clean; the tenant overwrites with v2; the |
| +// tag is still absent, so the door falls through to the result object and |
| +// answers clean for bytes nobody has read. That is the laundering the verdict |
| +// tag's own binding exists to stop, reintroduced by the mechanism meant to |
| +// close a gap in it. |
| +func TestResultObjectIsVersionBound(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", resultFixture("uploads/x", "clean", "v1")); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + got, err := ReadScanResult(ctx, s, "bab", "titogo", "uploads/x", "v1") |
| + if err != nil || got == nil || got.Verdict != "clean" { |
| + t.Fatalf("read at v1 = %+v, %v", got, err) |
| + } |
| + |
| + // The object is overwritten. The result must not answer for the new bytes. |
| + got, err = ReadScanResult(ctx, s, "bab", "titogo", "uploads/x", "v2") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got != nil { |
| + t.Fatalf("a result written for v1 answered for v2: %+v", got) |
| + } |
| +} |
| + |
| +// The second write for a key is a REPLACE. release.Store.Put has only |
| +// create-only and replace, so a create-only second write returns |
| +// ErrCASConflict and silently drops the newer verdict — leaving the door on |
| +// "pending" until some sweep happens to tag the object. |
| +func TestResultObjectReplacesOnSecondWrite(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", resultFixture("uploads/x", "clean", "v1")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", resultFixture("uploads/x", "infected", "v2")); err != nil { |
| + t.Fatalf("second write: %v", err) |
| + } |
| + |
| + got, err := ReadScanResult(ctx, s, "bab", "titogo", "uploads/x", "v2") |
| + if err != nil || got == nil { |
| + t.Fatalf("read at v2 = %+v, %v", got, err) |
| + } |
| + if got.Verdict != "infected" { |
| + t.Fatalf("verdict = %q, want the v2 verdict — the newer one was dropped", got.Verdict) |
| + } |
| +} |
| + |
| +// Absent is a state, not an error. |
| +func TestReadScanResultAbsent(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + |
| + got, err := ReadScanResult(ctx, s, "bab", "titogo", "uploads/never", "v1") |
| + if err != nil || got != nil { |
| + t.Fatalf("= %+v, %v", got, err) |
| + } |
| +} |
| + |
| +// An unknown schema refuses rather than being read as a verdict. Same rule as |
| +// the record and the sidecar, and for the same reason: this one can be the |
| +// only thing standing between an app and a payload. |
| +func TestScanResultSchemaRefusesUnknown(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + |
| + if _, err := s.Put(ctx, ScanResultKey("bab", "titogo", "uploads/x"), |
| + []byte(`{"schema":99,"key":"uploads/x","verdict":"clean","version":"v1"}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := ReadScanResult(ctx, s, "bab", "titogo", "uploads/x", "v1"); err == nil { |
| + t.Fatal("an unknown schema was read as a verdict") |
| + } |
| +} |
| + |
| +// The result lives under the APP's prefix, so PurgeApp removes it and the |
| +// name-claim gate can see it. A top-level results prefix would survive app |
| +// deletion invisibly — and, because materialize counts any key under an app |
| +// prefix and fails toward keeping the row, would resurrect the deleted app. |
| +func TestScanResultKeyIsUnderTheAppPrefix(t *testing.T) { |
| + got := ScanResultKey("bab", "titogo", "uploads/x") |
| + if want := release.AppPrefix("bab", "titogo"); len(got) <= len(want) || got[:len(want)] != want { |
| + t.Fatalf("key %q is not under %q", got, want) |
| + } |
| + // Distinct object keys must not collide. |
| + if ScanResultKey("bab", "titogo", "a") == ScanResultKey("bab", "titogo", "b") { |
| + t.Fatal("two object keys share a result object") |
| + } |
| + // The same key is stable, or the door could never find it again. |
| + if ScanResultKey("bab", "titogo", "a") != ScanResultKey("bab", "titogo", "a") { |
| + t.Fatal("result keys are not stable") |
| + } |
| +} |
| + |
| +// Deleted when a later scan DOES persist a verdict, so the fallback cannot |
| +// outlive the real answer. Deleting an absent one is not an error. |
| +func TestDeleteScanResult(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", resultFixture("uploads/x", "clean", "v1")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := DeleteScanResult(ctx, s, "bab", "titogo", "uploads/x"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", "uploads/x")); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("result survived deletion: %v", err) |
| + } |
| + if err := DeleteScanResult(ctx, s, "bab", "titogo", "uploads/x"); err != nil { |
| + t.Fatalf("deleting an absent result: %v", err) |
| + } |
| +} |