From 413084a9f927e75590c7c5d38cb9a2de40db161c Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Wed, 21 Oct 2026 16:55:12 -0400 Subject: [PATCH] feat(accounts): the object-storage connector diffs the current listing against one the host retains An object store answers only "what is there now". objectstore.FilesConnector versions a source by the digest of its listing (key, ETag, size) under its binding, the host retains the listing each version names in datasource_object_manifests (migration 46, tenant-owned, newest four per source), and Changes(from) diffs against it. It reaches the store through a Bucket interface the object-storage service's per-call binding will implement (codefly-dev/service-object-storage#108), and passes connectortest.RunFiles over a scripted bucket. It is not yet what the host syncs through: the provider stays registered non-conformant and runUploadSync keeps running. The upload descriptor now states source-scoped readers in the catalog. Refs #1211 Co-Authored-By: Claude Opus 5.5 (1M context) --- module/DATABASE_AUTHORITY.md | 2 +- module/services/accounts/AGENTS.md | 34 ++ .../pkg/adapters/datasource_handler_test.go | 5 + .../pkg/business/datasource_connectors.go | 5 +- .../pkg/datasource/connector/CONFORMANCE.md | 49 ++- .../pkg/datasource/objectstore/connector.go | 384 ++++++++++++++++++ .../objectstore/connector_conformance_test.go | 253 ++++++++++++ .../pkg/infra/postgres_object_manifests.go | 58 +++ .../infra/postgres_object_manifests_test.go | 79 ++++ .../pkg/infra/postgres_role_hardening_test.go | 8 +- .../code/pkg/relationcatalog/catalog.go | 1 + .../46_datasource_object_manifests.down.sql | 1 + .../46_datasource_object_manifests.up.sql | 23 ++ 13 files changed, 887 insertions(+), 15 deletions(-) create mode 100644 module/services/accounts/code/pkg/datasource/objectstore/connector.go create mode 100644 module/services/accounts/code/pkg/datasource/objectstore/connector_conformance_test.go create mode 100644 module/services/accounts/code/pkg/infra/postgres_object_manifests.go create mode 100644 module/services/accounts/code/pkg/infra/postgres_object_manifests_test.go create mode 100644 module/services/store/migrations/46_datasource_object_manifests.down.sql create mode 100644 module/services/store/migrations/46_datasource_object_manifests.up.sql diff --git a/module/DATABASE_AUTHORITY.md b/module/DATABASE_AUTHORITY.md index c1aa31212..2b5bfbee3 100644 --- a/module/DATABASE_AUTHORITY.md +++ b/module/DATABASE_AUTHORITY.md @@ -526,7 +526,7 @@ executable inventory and this table in the same change. | Scope | Relations | Required database boundary | |---|---|---| | `global` | `audit_event_types`, `bootstrap_state`, `data_retention_policies`, `datasource_credential_budgets`, `email_templates`, `feature_flags`, `identity_providers`, `module_envelopes`, `plan_entitlements`, `plans`, `platform_admins`, `policy_log_commits`, `policy_log_cursor`, `solution_delivery_documents`, `solution_generation_history`, `solution_host_bindings`, `solution_registrations`, `solution_targets` | No RLS; exact grants | -| `tenant` | `actor_chain_journal`, `actor_chain_revocations`, `api_keys`, `approval_decisions`, `approval_requests`, `audit_event_idempotency`, `audit_events`, `connector_credentials`, `dashboards`, `datasource_account_links`, `datasource_domains`, `datasource_group_bindings`, `codefly_effect_receipts`, `datasource_operation_attempts`, `datasource_oauth_consumed_states`, `datasource_source_operations`, `datasource_sources`, `delegation_grants`, `domain_events`, `entitlement_overrides`, `executable_artifact_approvals`, `github_app_installations`, `github_app_setups`, `installations`, `invitations`, `membership_integrity_findings`, `org_generic_settings`, `org_identity_providers`, `org_key_bindings`, `org_settings`, `organization_activations`, `organization_authorization_revisions`, `organization_members`, `organizations`, `principal_authorization_revisions`, `principals`, `record_shares`, `role_assignments`, `role_permissions`, `roles`, `scope_grants`, `scope_nodes`, `source_delegations`, `source_read_revisions`, `subscriptions`, `team_members`, `team_membership_quarantine`, `teams`, `usage_events`, `usage_totals`, `webhook_deliveries`, `webhook_subscriptions`, `work_context_replay` | Enabled and forced RLS with at least one policy | +| `tenant` | `actor_chain_journal`, `actor_chain_revocations`, `api_keys`, `approval_decisions`, `approval_requests`, `audit_event_idempotency`, `audit_events`, `connector_credentials`, `dashboards`, `datasource_account_links`, `datasource_domains`, `datasource_group_bindings`, `datasource_object_manifests`, `codefly_effect_receipts`, `datasource_operation_attempts`, `datasource_oauth_consumed_states`, `datasource_source_operations`, `datasource_sources`, `delegation_grants`, `domain_events`, `entitlement_overrides`, `executable_artifact_approvals`, `github_app_installations`, `github_app_setups`, `installations`, `invitations`, `membership_integrity_findings`, `org_generic_settings`, `org_identity_providers`, `org_key_bindings`, `org_settings`, `organization_activations`, `organization_authorization_revisions`, `organization_members`, `organizations`, `principal_authorization_revisions`, `principals`, `record_shares`, `role_assignments`, `role_permissions`, `roles`, `scope_grants`, `scope_nodes`, `source_delegations`, `source_read_revisions`, `subscriptions`, `team_members`, `team_membership_quarantine`, `teams`, `usage_events`, `usage_totals`, `webhook_deliveries`, `webhook_subscriptions`, `work_context_replay` | Enabled and forced RLS with at least one policy | | `user` | `client_authorization_codes`, `gdpr_requests`, `mfa_backup_codes`, `mfa_devices`, `mfa_login_transactions`, `notifications`, `onboarding_progress`, `resource_follows`, `sessions`, `user_consent_events`, `user_consent_preferences`, `user_identities`, `users`, `webauthn_ceremonies`, `webauthn_credentials` | Enabled and forced RLS with at least one policy | | `pre_auth` | `magic_links`, `waitlist_entries` | Enabled and forced RLS; fail-closed request policy, accessed only by the control-plane role | | `job` | `job_messages` | Enabled and forced RLS with at least one policy; no request relation grant — function-only scoped enqueue plus exact job-worker grants | diff --git a/module/services/accounts/AGENTS.md b/module/services/accounts/AGENTS.md index efe99b0ec..ad4d78b08 100644 --- a/module/services/accounts/AGENTS.md +++ b/module/services/accounts/AGENTS.md @@ -1173,3 +1173,37 @@ cause (v2 of that type; `datasourceAccessLostCodes` is the whole vocabulary a registry test holds the declaration to). Recovery is the ordinary one — a later snapshot clearing the degrade — which a repository made public again, or reconnected with a PAT or through the App, all reach. + +## Object storage is a client of the object-storage service, diffed against a listing the host retains + +The `upload` provider — "Object storage" to a person — reads a bucket under a +key prefix into a `collection` scope node. Its readers are source-scoped: a +bucket has no per-object access list, so the model to offer is one collection +per prefix, granted like any scope. It is registered non-conformant, so +`AdmitNewSource` refuses a new one; existing sources sync on `runUploadSync` +over the host's own S3 client (`pkg/datasource/objectstore/client.go`), daily +and on "Sync now", re-sending every object and closing a whole-folder sync with +the listing the store tombstones the rest against. `AddSource` refuses a +webhook secret for it: there is no push path (#1211 b). + +The owner's direction (2026-10-21) is that the host **re-uses the composed +object-storage service** rather than keep a store client of its own: the +service takes a per-call bucket binding with the credential handed per call and +owns the public-destination guard and the size caps +(codefly-dev/service-object-storage#108). What lives here is what is per-source +state: + +- **The retained listing is the cursor.** `objectstore.FilesConnector` versions + a source by the digest of its listing under its binding and diffs the current + listing against the one a version names. The host keeps those listings in + `datasource_object_manifests` (migration 46), tenant-owned — the control plane + has no grant — and pruned to the newest four per source by + `PutDatasourceObjectManifest`; a pruned or rebound version resyncs. Clause by + clause, in [CONFORMANCE.md](./code/pkg/datasource/connector/CONFORMANCE.md). +- **The connector reaches the store through `objectstore.Bucket`.** It lists the + prefix whole or fails (`ErrListingIncomplete` — a partial listing would read + every unlisted object as deleted) and reads a batch in one call at the listed + ETags. The implementation over the service is not written yet; until it is, + the connector is exercised only by the conformance suite over a scripted + bucket, the provider stays registered non-conformant, and `client.go` stays + the running engine. It is deleted when the service client lands, not extended. diff --git a/module/services/accounts/code/pkg/adapters/datasource_handler_test.go b/module/services/accounts/code/pkg/adapters/datasource_handler_test.go index afbf185fb..fcd137495 100644 --- a/module/services/accounts/code/pkg/adapters/datasource_handler_test.go +++ b/module/services/accounts/code/pkg/adapters/datasource_handler_test.go @@ -60,6 +60,11 @@ func TestDatasourceCatalog_ProjectsTheRegistry(t *testing.T) { t.Fatalf("%s entry = %v, want listed, non-conformant, with a gap, accepting no new source", key, p) } } + // A bucket has no per-object access list; the catalog says so before a + // person connects one. + if r := byKey[business.DatasourceProviderUpload].GetReadersModel(); r != gen.DatasourceReadersModel_DATASOURCE_READERS_MODEL_SOURCE_SCOPED { + t.Fatalf("object storage readers model = %v, want source-scoped", r) + } if datasourceCatalog(nil).GetProviders() != nil { t.Fatal("an unconfigured connector offers nothing") } diff --git a/module/services/accounts/code/pkg/business/datasource_connectors.go b/module/services/accounts/code/pkg/business/datasource_connectors.go index 2ba1db5fc..269fea9c3 100644 --- a/module/services/accounts/code/pkg/business/datasource_connectors.go +++ b/module/services/accounts/code/pkg/business/datasource_connectors.go @@ -74,7 +74,10 @@ func (s *Service) newDatasourceConnectorRegistry() *connector.Registry { Description: "An S3-compatible bucket; objects under a prefix are pulled on sync.", Interface: connector.InterfaceFiles, CredentialModes: []connector.CredentialMode{connector.CredentialStaticSecret}, - Gap: gapUploadDatasource, + // A bucket has no per-object access list, so the catalog says before a + // person asks that every object is readable by the collection's readers. + Readers: connector.ReadersSourceScoped, + Gap: gapUploadDatasource, })) return r } diff --git a/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md b/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md index f175f11a8..17ff92a9c 100644 --- a/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md +++ b/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md @@ -54,13 +54,41 @@ closes a whole-folder sync. | 6 budget and backpressure | No declared `Budget`, and the connector is not metered by `datasource_credential_budgets`, so an object-storage sync cannot yield to a person's sync the way a GitHub one does. | | 8 bulk only | One `GetObject` per object. The listing is one call; the fetches are not. | -**What closing it takes.** A `FilesConnector` over S3 whose version is a digest -of the listing (key → ETag) and whose `Changes(from)` diffs the current listing -against the one `from` names — which needs the previous listing to be retained, -since S3 has no "what changed since" API. That is a store-shaped change, not -just a connector: the host would have to keep the last manifest per source. Not -done here because it is a design question about where that manifest lives, not a -gap in a flow that already exists. +**What closing it takes, and how far it has got.** The connector that closes +clause 4 exists and passes `connectortest.RunFiles` +(`pkg/datasource/objectstore/connector_conformance_test.go`); it is not yet what +the host syncs through, so the registration above is unchanged and every +object-storage source still runs on `runUploadSync`. + +- **The cursor is a retained listing.** An object store answers only "what is + there now", so `objectstore.FilesConnector` versions a source by a digest of + its listing (key, ETag, size) under its binding (endpoint, region, bucket, + prefix), and the host keeps the listing each version names in + `datasource_object_manifests` (migration 46; tenant-owned, the newest four + per source). `Changes(from)` diffs the current listing against the retained + one; a version whose listing was pruned, or that was listed under another + bucket or prefix, is `ErrResyncRequired`. A key that left and a key that + arrived with the same ETag and size are reported as one MOVED change — a + store has no rename, and the same bytes at a new key is what a move is. +- **A fetch is at a version.** `FetchFiles` serves an object only at the ETag + the version's listing names; an object rewritten or deleted since is + `ErrItemNotFound`, never content the version did not have. +- **The store is reached through a `Bucket`, not spoken to.** The connector + signs, dials and guards nothing: listing and reading are the composed + object-storage service's, under a per-call binding with the credential the + host decrypts (codefly-dev/service-object-storage#108). `client.go` in this + package is the engine `runUploadSync` still uses; it is deleted when the + `Bucket` over the service lands, not finished. +- **Clause 8 rests on the service.** The suite's scripted bucket serves a batch + in one read, which is the contract `Bucket.Read` states. Whether the + service's binding serves a batch in one call or one call per object decides + whether that holds against a real store; until the client exists it is the + stated requirement, not a measured fact. + +What remains before the provider is registered conformant: the `Bucket` over the +service, `runUploadSync` replaced by the envelope path, the connector registered +through `connector.Budgeted` so it is metered by `datasource_credential_budgets` +(clause 6), and the gap sentence above deleted. ## crawler — documentation website @@ -146,6 +174,7 @@ not qualify database commit/rollback or cross-process serialization. | api | no | yes (records) | no | no | no | Object storage is the nearest: it is already on an admissible interface and -already applies deletions, and one cursor stands between it and a second -conformant implementation. That is what would demonstrate the model generalises, -and it is the one worth doing next. +already applies deletions, and its cursor is built and passes the suite against +a scripted store. What stands between it and a second conformant implementation +is the client of the object-storage service the host syncs through. That is +what would demonstrate the model generalises. diff --git a/module/services/accounts/code/pkg/datasource/objectstore/connector.go b/module/services/accounts/code/pkg/datasource/objectstore/connector.go new file mode 100644 index 000000000..c5699f717 --- /dev/null +++ b/module/services/accounts/code/pkg/datasource/objectstore/connector.go @@ -0,0 +1,384 @@ +package objectstore + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "sort" + "strings" + "time" + + "accounts/pkg/datasource/connector" + accountsv1 "accounts/pkg/gen/saas/accounts/v1" +) + +// The object-storage files connector, on the envelope. An object store has no +// "what changed since": it answers only "what is there now". So the version of +// a source is a digest of its listing (key → ETag → size), the host retains the +// listing each version names (Manifests), and Changes diffs the current listing +// against the retained one. The listing and the reads go through a Bucket, which +// is the composed object-storage service under the source's binding +// (codefly-dev/service-object-storage#108): this connector signs nothing, dials +// nothing and guards nothing itself. + +// ConnectorKey is the provider key object-storage sources are stored under. +const ConnectorKey = "upload" + +// Budget limits. An object past MaxItemBytes is refused, never truncated. +const ( + MaxObjectsPerBatch = 1000 + MaxBatchBytes = 64 << 20 + MaxObjectBytes = 5 << 20 +) + +// estimatedThrottleWindow is the reset a throttled call is given when the store +// does not say when its limit resets. +const estimatedThrottleWindow = time.Minute + +// SourceConfig is where one source reads: its store, bucket and key prefix. The +// credential is not here; the host hands it to the Bucket per call. +type SourceConfig struct { + Endpoint string + Region string + Bucket string + Prefix string +} + +// binding names the configuration a version was listed under, so a version of +// one bucket or prefix is never diffed against, or read from, another. +func (c SourceConfig) binding() string { + sum := sha256.Sum256([]byte(c.Endpoint + "\x00" + c.Region + "\x00" + c.Bucket + "\x00" + c.Prefix)) + return hex.EncodeToString(sum[:8]) +} + +// Item is one object of a listing: its full key (the item's stable id), its +// ETag (the item's version) and its exact size in bytes. +type Item struct { + Key string `json:"key"` + ETag string `json:"etag"` + Size int64 `json:"size"` +} + +// ErrListingIncomplete: the store could not list the whole prefix (it truncated +// the listing, or the listing passed the store's cap). A partial listing is never +// a version: it would read every unlisted object as deleted. +var ErrListingIncomplete = errors.New("objectstore: the store did not list the whole prefix") + +// ErrObjectChanged: an object named at an ETag no longer has it (it was +// rewritten or deleted after it was listed). +var ErrObjectChanged = errors.New("objectstore: the object is no longer at the listed version") + +// Bucket is the store as the connector reads it, for one source. It reports a +// throttle as *connector.RateLimitedError and a refused credential as +// connector.ErrCredentialRefused. +type Bucket interface { + // List returns every object under the source's prefix, or + // ErrListingIncomplete; never a partial listing. + List(ctx context.Context) ([]Item, error) + // Read streams the named objects in one call, in order, each only at its + // listed ETag (ErrObjectChanged otherwise), with a reader of exactly its size. + Read(ctx context.Context, items []Item, visit func(item Item, contentType string, content io.Reader) error) error +} + +// BucketFunc resolves the Bucket of one source, per call: the host decrypts the +// source's credential and binds it, and the connector never sees it. +type BucketFunc func(ctx context.Context, src connector.Source) (Bucket, error) + +// Manifests is the host's retained listings, per source and version. Put is +// idempotent; Get reports false for a version that was never retained or has +// since been pruned. +type Manifests interface { + Put(ctx context.Context, src connector.Source, version string, items []Item) error + Get(ctx context.Context, src connector.Source, version string) ([]Item, bool, error) +} + +// FilesConnector is the object-storage connector. +type FilesConnector struct { + bucket BucketFunc + manifests Manifests + budget connector.Budget + now func() time.Time +} + +// ConnectorOption configures a FilesConnector. +type ConnectorOption func(*FilesConnector) + +// WithMaxItemBytes lowers the per-object limit, for a test that must script an +// over-limit object cheaply. It never raises it past MaxObjectBytes. +func WithMaxItemBytes(n int64) ConnectorOption { + return func(c *FilesConnector) { + if n > 0 && n < c.budget.MaxItemBytes { + c.budget.MaxItemBytes = n + } + } +} + +// NewFilesConnector returns the object-storage connector over bucket, retaining +// listings in manifests. +func NewFilesConnector(bucket BucketFunc, manifests Manifests, opts ...ConnectorOption) *FilesConnector { + c := &FilesConnector{ + bucket: bucket, + manifests: manifests, + budget: connector.Budget{ + MaxItemsPerCall: MaxObjectsPerBatch, MaxBytesPerCall: MaxBatchBytes, MaxItemBytes: MaxObjectBytes, + // One operation is one listing or one batched read. A store meters + // requests per second, far above this; the quota is what the host + // spends on a tenant's bucket, with a fifth kept for a person's sync. + OperationsPerWindow: 1200, Window: time.Hour, BackgroundSharePercent: 80, + }, + now: time.Now, + } + for _, opt := range opts { + opt(c) + } + return c +} + +// Descriptor is the connector's catalog entry. +func (c *FilesConnector) Descriptor() connector.Descriptor { + return connector.Descriptor{ + Key: ConnectorKey, + DisplayName: "Object storage", + Description: "A bucket in an object store; the objects under a key prefix are pulled on sync, and only what changed is re-sent.", + Interface: connector.InterfaceFiles, + ConfigFields: []connector.ConfigField{ + {Key: "endpoint", DisplayName: "Endpoint", Help: "The store's https base URL.", Required: true}, + {Key: "region", DisplayName: "Region", Help: "The bucket's region, e.g. us-east-1.", Required: true}, + {Key: "bucket", DisplayName: "Bucket", Required: true}, + {Key: "prefix", DisplayName: "Prefix", Help: "Key prefix to read under; empty means the whole bucket.", Required: false}, + {Key: "access_key_id", DisplayName: "Access key ID", Required: true}, + }, + CredentialModes: []connector.CredentialMode{connector.CredentialStaticSecret}, + // A bucket has no per-object access list: every object is readable by + // whoever may read the source's collection. One collection per prefix. + Readers: connector.ReadersSourceScoped, + Budget: c.budget, + Conformant: true, + } +} + +func (c *FilesConnector) open(ctx context.Context, src connector.Source) (SourceConfig, Bucket, error) { + cfg, ok := src.Config.(SourceConfig) + if !ok || cfg.Bucket == "" { + return SourceConfig{}, nil, fmt.Errorf("objectstore connector: source %s has no object-storage configuration", src.ID) + } + if err := ctx.Err(); err != nil { + return SourceConfig{}, nil, err + } + bucket, err := c.bucket(ctx, src) + if err != nil { + return SourceConfig{}, nil, c.envelopeError(err) + } + return cfg, bucket, nil +} + +// list reads the current listing and its version. Folder markers (keys ending +// in "/") are not files and are left out. +func (c *FilesConnector) list(ctx context.Context, cfg SourceConfig, bucket Bucket) (string, map[string]Item, error) { + items, err := bucket.List(ctx) + if err != nil { + return "", nil, c.envelopeError(err) + } + listed := make(map[string]Item, len(items)) + for _, it := range items { + if !strings.HasSuffix(it.Key, "/") { + listed[it.Key] = it + } + } + return versionOf(cfg, listed), listed, nil +} + +// versionOf is the listing's digest under its binding: the binding's digest, +// then the digest of every key, ETag and size in key order. +func versionOf(cfg SourceConfig, listed map[string]Item) string { + h := sha256.New() + for _, key := range sortedKeys(listed) { + it := listed[key] + fmt.Fprintf(h, "%s\x00%s\x00%d\n", it.Key, it.ETag, it.Size) + } + return cfg.binding() + "." + hex.EncodeToString(h.Sum(nil)) +} + +func sortedKeys(m map[string]Item) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +// Version lists the prefix and digests it. An object store answers no cheaper +// question than its listing, so this costs one listing call. +func (c *FilesConnector) Version(ctx context.Context, src connector.Source) (string, error) { + cfg, bucket, err := c.open(ctx, src) + if err != nil { + return "", err + } + version, _, err := c.list(ctx, cfg, bucket) + return version, err +} + +// Changes diffs the current listing against the retained listing from names, +// and retains the current one under its version. An empty from is a complete +// snapshot. A from that was listed under another binding, or whose listing is +// no longer retained, requires a resync. +// +// An object store has no rename: a key that disappeared and a key that appeared +// with the same ETag and size are the same content moved, and are reported as +// one MOVED change, paired in key order. +func (c *FilesConnector) Changes(ctx context.Context, src connector.Source, from string) (connector.ChangeSet, error) { + cfg, bucket, err := c.open(ctx, src) + if err != nil { + return connector.ChangeSet{}, err + } + var previous map[string]Item + if from != "" { + if !strings.HasPrefix(from, cfg.binding()+".") { + return connector.ChangeSet{}, connector.ErrResyncRequired + } + items, ok, err := c.manifests.Get(ctx, src, from) + if err != nil { + return connector.ChangeSet{}, err + } + if !ok { + return connector.ChangeSet{}, connector.ErrResyncRequired + } + previous = make(map[string]Item, len(items)) + for _, it := range items { + previous[it.Key] = it + } + } + version, listed, err := c.list(ctx, cfg, bucket) + if err != nil { + return connector.ChangeSet{}, err + } + set := connector.ChangeSet{SourceID: src.ID, From: from, To: version, Complete: from == ""} + if version == from { + return set, nil + } + current := make([]Item, 0, len(listed)) + for _, key := range sortedKeys(listed) { + current = append(current, listed[key]) + } + if err := c.manifests.Put(ctx, src, version, current); err != nil { + return connector.ChangeSet{}, err + } + readers := connector.ApplySourcePolicy(src, connector.SourceScoped()) + change := func(it Item, kind connector.ChangeKind) connector.Change { + return connector.Change{Key: connector.KeyFor(src, it.Key), Kind: kind, ItemVersion: it.ETag, Size: it.Size, Locator: it.Key, Readers: readers} + } + var added, deleted []Item + for _, it := range current { + prev, ok := previous[it.Key] + switch { + case !ok: + added = append(added, it) + case prev.ETag != it.ETag || prev.Size != it.Size: + set.Changes = append(set.Changes, change(it, connector.ChangeModified)) + } + } + for _, key := range sortedKeys(previous) { + if _, ok := listed[key]; !ok { + deleted = append(deleted, previous[key]) + } + } + moved := make(map[string]bool) + for _, it := range added { + kind, prevKey := connector.ChangeAdded, "" + for _, gone := range deleted { + if !moved[gone.Key] && gone.ETag == it.ETag && gone.Size == it.Size { + moved[gone.Key], kind, prevKey = true, connector.ChangeMoved, gone.Key + break + } + } + ch := change(it, kind) + ch.PreviousItemID = prevKey + set.Changes = append(set.Changes, ch) + } + for _, gone := range deleted { + if !moved[gone.Key] { + set.Changes = append(set.Changes, connector.Change{Key: connector.KeyFor(src, gone.Key), Kind: connector.ChangeDeleted, Locator: gone.Key}) + } + } + return set, nil +} + +// FetchFiles reads the named objects as the retained listing of version names +// them, in one batched read. An object rewritten or deleted since that listing +// is not at that version any more, and is ErrItemNotFound. +func (c *FilesConnector) FetchFiles(ctx context.Context, src connector.Source, version string, refs []connector.FileRef, visit func(connector.File, io.Reader) error) error { + if len(refs) == 0 || len(refs) > c.budget.MaxItemsPerCall { + return fmt.Errorf("%w: a batch names 1 to %d objects", connector.ErrBatchTooLarge, c.budget.MaxItemsPerCall) + } + seen := make(map[string]bool, len(refs)) + for _, ref := range refs { + if ref.ItemID == "" || seen[ref.ItemID] { + return errors.New("objectstore connector: every object of a batch names a distinct item") + } + seen[ref.ItemID] = true + } + cfg, bucket, err := c.open(ctx, src) + if err != nil { + return err + } + if !strings.HasPrefix(version, cfg.binding()+".") { + return connector.ErrVersionNotFound + } + retained, ok, err := c.manifests.Get(ctx, src, version) + if err != nil { + return err + } + if !ok { + return connector.ErrVersionNotFound + } + byKey := make(map[string]Item, len(retained)) + for _, it := range retained { + byKey[it.Key] = it + } + items := make([]Item, len(refs)) + var total int64 + for i, ref := range refs { + it, ok := byKey[ref.ItemID] + if !ok || (ref.ItemVersion != "" && ref.ItemVersion != it.ETag) { + return connector.ErrItemNotFound + } + if it.Size > c.budget.MaxItemBytes { + return fmt.Errorf("%w: an object is over %d bytes", connector.ErrItemTooLarge, c.budget.MaxItemBytes) + } + items[i], total = it, total+it.Size + } + if total > c.budget.MaxBytesPerCall { + return fmt.Errorf("%w: the batch is %d bytes, over %d; request fewer objects", connector.ErrBatchTooLarge, total, c.budget.MaxBytesPerCall) + } + readers := connector.ApplySourcePolicy(src, connector.SourceScoped()) + err = bucket.Read(ctx, items, func(it Item, contentType string, content io.Reader) error { + return visit(connector.File{ + Provenance: &accountsv1.DatasourceProvenance{ + SourceId: src.ID, OrgId: src.OrgID, BoundaryNodeId: src.BoundaryNodeID, + Version: version, ItemId: it.Key, ItemVersion: it.ETag, + }, + Locator: it.Key, + ContentType: contentType, + Size: it.Size, + Readers: readers, + }, content) + }) + if errors.Is(err, ErrObjectChanged) { + return fmt.Errorf("%w: %w", connector.ErrItemNotFound, err) + } + return c.envelopeError(err) +} + +// envelopeError gives a throttle the store did not time a reset, keeping every +// other error, typed by the Bucket, as it is. +func (c *FilesConnector) envelopeError(err error) error { + var limited *connector.RateLimitedError + if errors.As(err, &limited) && limited.ResetAt.IsZero() { + limited.ResetAt, limited.Estimated = c.now().Add(estimatedThrottleWindow), true + } + return err +} diff --git a/module/services/accounts/code/pkg/datasource/objectstore/connector_conformance_test.go b/module/services/accounts/code/pkg/datasource/objectstore/connector_conformance_test.go new file mode 100644 index 000000000..568c485ff --- /dev/null +++ b/module/services/accounts/code/pkg/datasource/objectstore/connector_conformance_test.go @@ -0,0 +1,253 @@ +package objectstore + +import ( + "context" + "crypto/md5" + "encoding/hex" + "errors" + "fmt" + "io" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "accounts/pkg/datasource/connector" + "accounts/pkg/datasource/connector/connectortest" +) + +// TestFilesConnectorConformance runs the datasource conformance suite against +// the object-storage connector, over a scripted bucket that requires a +// sentinel credential and counts every call: one per listing, one per batched +// read, which is the contract Bucket states. +func TestFilesConnectorConformance(t *testing.T) { + connectortest.RunFiles(t, newConformanceFixture) +} + +const conformanceCredential = "sentinel-secret-key-4c1d" +const conformanceMaxItemBytes = 4 << 10 + +var conformanceConfig = SourceConfig{Endpoint: "https://store.example.com", Region: "us-east-1", Bucket: "acme-docs", Prefix: "kb/"} + +type conformanceFixture struct { + store *fakeStore + manifests *memoryManifests + conn *FilesConnector + sources int +} + +func newConformanceFixture(t *testing.T) connectortest.FilesFixture { + f := &conformanceFixture{store: &fakeStore{objects: map[string]string{}}, manifests: &memoryManifests{}} + f.conn = NewFilesConnector(func(context.Context, connector.Source) (Bucket, error) { + // The host resolves the credential per call; here it is the sentinel. + return &fakeBucket{store: f.store, credential: conformanceCredential}, nil + }, f.manifests, WithMaxItemBytes(conformanceMaxItemBytes)) + return f +} + +func (f *conformanceFixture) Connector() connector.FilesConnector { return f.conn } + +func (f *conformanceFixture) Source(t *testing.T, org string) connector.Source { + f.sources++ + return connector.Source{ + ID: fmt.Sprintf("aaaaaaaa-0000-0000-0000-%012d", f.sources), + OrgID: org, + BoundaryNodeID: fmt.Sprintf("bbbbbbbb-0000-0000-0000-%012d", f.sources), + Config: conformanceConfig, + } +} + +// Apply writes the provider's next state. A store has no rename: a move is the +// object written at its new key and removed from its old one. +func (f *conformanceFixture) Apply(t *testing.T, puts map[string]string, deletes []string, moves map[string]string) { + f.store.mu.Lock() + defer f.store.mu.Unlock() + for key, content := range puts { + f.store.objects[key] = content + } + for _, key := range deletes { + delete(f.store.objects, key) + } + for from, to := range moves { + f.store.objects[to] = f.store.objects[from] + delete(f.store.objects, from) + } +} + +// Rewrite: a bucket has no history to rewrite. What makes a version +// undiffable is that the listing it names is no longer retained, so the +// fixture drops every retained listing. +func (f *conformanceFixture) Rewrite(t *testing.T) { f.manifests.forget() } + +func (f *conformanceFixture) ProviderCalls() int64 { return f.store.calls.Load() } + +func (f *conformanceFixture) RateLimit(t *testing.T) func() { + f.store.throttled.Store(true) + return func() { f.store.throttled.Store(false) } +} + +func (f *conformanceFixture) Credential() string { return conformanceCredential } + +func (f *conformanceFixture) CredentialPresented() bool { return f.store.authenticated.Load() } + +func (f *conformanceFixture) MaxItemBytes() int64 { return conformanceMaxItemBytes } + +// fakeStore is one bucket's objects, shared by every source of a fixture. +type fakeStore struct { + mu sync.Mutex + objects map[string]string + calls atomic.Int64 + throttled atomic.Bool + authenticated atomic.Bool +} + +func etagOf(content string) string { + sum := md5.Sum([]byte(content)) + return `"` + hex.EncodeToString(sum[:]) + `"` +} + +// fakeBucket is the store as one call's credential sees it. +type fakeBucket struct { + store *fakeStore + credential string +} + +var errDenied = errors.New("fake store: access denied") + +func (b *fakeBucket) call(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + b.store.calls.Add(1) + if b.credential != conformanceCredential { + return fmt.Errorf("%w: %w", connector.ErrCredentialRefused, errDenied) + } + b.store.authenticated.Store(true) + if b.store.throttled.Load() { + // SlowDown carries no reset; the connector estimates one. + return &connector.RateLimitedError{Scope: "credential", Cause: errors.New("fake store: SlowDown")} + } + return nil +} + +func (b *fakeBucket) List(ctx context.Context) ([]Item, error) { + if err := b.call(ctx); err != nil { + return nil, err + } + b.store.mu.Lock() + defer b.store.mu.Unlock() + items := make([]Item, 0, len(b.store.objects)) + for key, content := range b.store.objects { + items = append(items, Item{Key: key, ETag: etagOf(content), Size: int64(len(content))}) + } + return items, nil +} + +func (b *fakeBucket) Read(ctx context.Context, items []Item, visit func(Item, string, io.Reader) error) error { + if err := b.call(ctx); err != nil { + return err + } + b.store.mu.Lock() + contents := make([]string, len(items)) + for i, it := range items { + content, ok := b.store.objects[it.Key] + if !ok || etagOf(content) != it.ETag { + b.store.mu.Unlock() + return ErrObjectChanged + } + contents[i] = content + } + b.store.mu.Unlock() + for i, it := range items { + if err := visit(it, "text/plain; charset=utf-8", strings.NewReader(contents[i])); err != nil { + return err + } + } + return nil +} + +// memoryManifests retains every listing until forget. +type memoryManifests struct { + mu sync.Mutex + kept map[string][]Item +} + +func (m *memoryManifests) Put(_ context.Context, src connector.Source, version string, items []Item) error { + m.mu.Lock() + defer m.mu.Unlock() + if m.kept == nil { + m.kept = map[string][]Item{} + } + m.kept[src.ID+"/"+version] = append([]Item(nil), items...) + return nil +} + +func (m *memoryManifests) Get(_ context.Context, src connector.Source, version string) ([]Item, bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + items, ok := m.kept[src.ID+"/"+version] + return items, ok, nil +} + +func (m *memoryManifests) forget() { + m.mu.Lock() + defer m.mu.Unlock() + m.kept = nil +} + +// TestChangesResyncAcrossABinding holds a version to the bucket and prefix it +// was listed under: after the source is pointed elsewhere, its old version is +// neither diffed against nor read from. +func TestChangesResyncAcrossABinding(t *testing.T) { + f := newConformanceFixture(t).(*conformanceFixture) + f.Apply(t, map[string]string{"kb/a.md": "alpha"}, nil, nil) + src := f.Source(t, "11111111-1111-1111-1111-111111111111") + ctx := context.Background() + a, err := f.conn.Changes(ctx, src, "") + if err != nil { + t.Fatal(err) + } + moved := conformanceConfig + moved.Prefix = "other/" + src.Config = moved + if _, err := f.conn.Changes(ctx, src, a.To); !errors.Is(err, connector.ErrResyncRequired) { + t.Fatalf("changes from a version of another prefix = %v, want ErrResyncRequired", err) + } + err = f.conn.FetchFiles(ctx, src, a.To, []connector.FileRef{{ItemID: "kb/a.md"}}, func(connector.File, io.Reader) error { return nil }) + if !errors.Is(err, connector.ErrVersionNotFound) { + t.Fatalf("fetching a version of another prefix = %v, want ErrVersionNotFound", err) + } +} + +// TestFetchRefusesARewrittenObject: an object rewritten after it was listed is +// not at the listed version any more, so a fetch at that version refuses it +// rather than serve content the version never had. +func TestFetchRefusesARewrittenObject(t *testing.T) { + f := newConformanceFixture(t).(*conformanceFixture) + f.Apply(t, map[string]string{"kb/a.md": "alpha"}, nil, nil) + src := f.Source(t, "11111111-1111-1111-1111-111111111111") + ctx := context.Background() + a, err := f.conn.Changes(ctx, src, "") + if err != nil { + t.Fatal(err) + } + f.Apply(t, map[string]string{"kb/a.md": "alpha, rewritten"}, nil, nil) + err = f.conn.FetchFiles(ctx, src, a.To, []connector.FileRef{{ItemID: "kb/a.md"}}, func(connector.File, io.Reader) error { return nil }) + if !errors.Is(err, connector.ErrItemNotFound) { + t.Fatalf("fetching an object rewritten since its listing = %v, want ErrItemNotFound", err) + } +} + +// TestRateLimitIsEstimatedWhenTheStoreGivesNoReset: a store's throttle names +// no reset, so the connector gives it one rather than a zero time. +func TestRateLimitIsEstimatedWhenTheStoreGivesNoReset(t *testing.T) { + f := newConformanceFixture(t).(*conformanceFixture) + clear := f.RateLimit(t) + defer clear() + _, err := f.conn.Changes(context.Background(), f.Source(t, "11111111-1111-1111-1111-111111111111"), "") + var limited *connector.RateLimitedError + if !errors.As(err, &limited) || !limited.Estimated || !limited.ResetAt.After(time.Now()) { + t.Fatalf("a throttle without a reset = %v, want an estimated reset in the future", err) + } +} diff --git a/module/services/accounts/code/pkg/infra/postgres_object_manifests.go b/module/services/accounts/code/pkg/infra/postgres_object_manifests.go new file mode 100644 index 000000000..6d3abdbf8 --- /dev/null +++ b/module/services/accounts/code/pkg/infra/postgres_object_manifests.go @@ -0,0 +1,58 @@ +package infra + +import ( + "context" + "encoding/json" + "errors" + + "github.com/jackc/pgx/v5" + + "accounts/pkg/datasource/objectstore" +) + +// retainedObjectManifests is how many listings a source keeps: the one its +// cursor names, the one a sync in flight just listed, and slack for a sync that +// listed and then failed before the cursor moved. A version past it resyncs. +const retainedObjectManifests = 4 + +// PutDatasourceObjectManifest retains one listing of an object-storage source +// under its version, inside the source's org transaction, and prunes all but +// the newest retained listings. A version is the listing's own digest, so a +// second put of it only marks it newest. +func (s *PostgresStore) PutDatasourceObjectManifest(ctx context.Context, org, source, version string, objects []objectstore.Item) error { + if objects == nil { + objects = []objectstore.Item{} + } + encoded, err := json.Marshal(objects) + if err != nil { + return err + } + q := s.getQueryExecutor(ctx) + if _, err := q.Exec(ctx, `INSERT INTO datasource_object_manifests (org_id,source_id,version,objects,retained_at) VALUES ($1,$2,$3,$4,clock_timestamp()) + ON CONFLICT (org_id,source_id,version) DO UPDATE SET retained_at = clock_timestamp()`, org, source, version, encoded); err != nil { + return err + } + _, err = q.Exec(ctx, `DELETE FROM datasource_object_manifests WHERE org_id=$1 AND source_id=$2 AND version NOT IN ( + SELECT version FROM datasource_object_manifests WHERE org_id=$1 AND source_id=$2 ORDER BY retained_at DESC, version LIMIT $3)`, + org, source, retainedObjectManifests) + return err +} + +// GetDatasourceObjectManifest returns the listing a source retained under +// version, and false when it never retained one or has pruned it. +func (s *PostgresStore) GetDatasourceObjectManifest(ctx context.Context, org, source, version string) ([]objectstore.Item, bool, error) { + var encoded []byte + err := s.getQueryExecutor(ctx).QueryRow(ctx, `SELECT objects FROM datasource_object_manifests WHERE org_id=$1 AND source_id=$2 AND version=$3`, + org, source, version).Scan(&encoded) + if errors.Is(err, pgx.ErrNoRows) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + var objects []objectstore.Item + if err := json.Unmarshal(encoded, &objects); err != nil { + return nil, false, err + } + return objects, true, nil +} diff --git a/module/services/accounts/code/pkg/infra/postgres_object_manifests_test.go b/module/services/accounts/code/pkg/infra/postgres_object_manifests_test.go new file mode 100644 index 000000000..7658a564f --- /dev/null +++ b/module/services/accounts/code/pkg/infra/postgres_object_manifests_test.go @@ -0,0 +1,79 @@ +//go:build !pure + +package infra_test + +import ( + "context" + "fmt" + "strings" + "testing" + + "accounts/pkg/datasource/objectstore" + + "github.com/stretchr/testify/require" +) + +func manifestVersion(i int) string { + return fmt.Sprintf("%016x.%064x", 0xabc, i) +} + +// TestPostgresDatasourceObjectManifestsRetainTheNewest pins the retained +// listings an object-storage source diffs against: a listing round-trips under +// its version, a source keeps only its newest few, a re-put of a version makes +// it newest again, and another organization reads none of them. +func TestPostgresDatasourceObjectManifestsRetainTheNewest(t *testing.T) { + userID := seedUser(t) + orgID := seedOrg(t, userID) + sourceID := seedDatasourceSource(t, orgID) + listing := []objectstore.Item{{Key: "kb/a.md", ETag: `"e1"`, Size: 5}, {Key: "kb/b.md", ETag: `"e2"`, Size: 7}} + + put := func(i int) { + require.NoError(t, testStore.WithOrgTx(testCtx, orgID, func(ctx context.Context) error { + return testStore.PutDatasourceObjectManifest(ctx, orgID, sourceID, manifestVersion(i), listing) + })) + } + get := func(org string, i int) ([]objectstore.Item, bool) { + var items []objectstore.Item + var ok bool + require.NoError(t, testStore.WithOrgTx(testCtx, org, func(ctx context.Context) error { + var err error + items, ok, err = testStore.GetDatasourceObjectManifest(ctx, orgID, sourceID, manifestVersion(i)) + return err + })) + return items, ok + } + + put(0) + items, ok := get(orgID, 0) + require.True(t, ok) + require.Equal(t, listing, items) + + for i := 1; i <= 3; i++ { + put(i) + } + put(0) // re-listed: newest again, so version 1 is now the oldest + put(4) + _, ok = get(orgID, 1) + require.False(t, ok, "the oldest listing past the retained count is pruned") + for _, i := range []int{0, 2, 3, 4} { + _, ok = get(orgID, i) + require.True(t, ok, "listing %d is among the newest and must be retained", i) + } + + otherOrg := seedOrg(t, seedUser(t)) + _, ok = get(otherOrg, 4) + require.False(t, ok, "another organization must not read a source's listings") +} + +// TestPostgresDatasourceObjectManifestRefusesAMalformedVersion: the version is +// the connector's digest, never a free string. +func TestPostgresDatasourceObjectManifestRefusesAMalformedVersion(t *testing.T) { + userID := seedUser(t) + orgID := seedOrg(t, userID) + sourceID := seedDatasourceSource(t, orgID) + err := testStore.WithOrgTx(testCtx, orgID, func(ctx context.Context) error { + return testStore.PutDatasourceObjectManifest(ctx, orgID, sourceID, "not-a-version", nil) + }) + require.Error(t, err) + require.True(t, strings.Contains(err.Error(), "check"), "want the version CHECK to refuse it, got %v", err) +} diff --git a/module/services/accounts/code/pkg/infra/postgres_role_hardening_test.go b/module/services/accounts/code/pkg/infra/postgres_role_hardening_test.go index ab73fe465..a83194110 100644 --- a/module/services/accounts/code/pkg/infra/postgres_role_hardening_test.go +++ b/module/services/accounts/code/pkg/infra/postgres_role_hardening_test.go @@ -85,6 +85,7 @@ var appTenantRelationPrivileges = map[string]relationPrivileges{ "installations": {selectRows: true, insertRows: true, updateRows: true, deleteRows: true}, "datasource_sources": {selectRows: true, insertRows: true, updateRows: true, deleteRows: true}, "datasource_source_operations": {selectRows: true, insertRows: true, updateRows: true, deleteRows: true}, + "datasource_object_manifests": {selectRows: true, insertRows: true, updateRows: true, deleteRows: true}, "datasource_operation_attempts": {selectRows: true, insertRows: true, deleteRows: true}, "codefly_effect_receipts": {selectRows: true, insertRows: true, deleteRows: true}, "datasource_oauth_consumed_states": {selectRows: true, insertRows: true, deleteRows: true}, @@ -371,11 +372,12 @@ func TestControlPlaneRelationGrantsAreExact(t *testing.T) { if relation == "executable_artifact_approvals" { want = relationPrivileges{} } - // Callable-source declarations, OAuth state, attempts and receipts - // are tenant-only. The control plane must not read or mutate them. + // Callable-source declarations, OAuth state, attempts, receipts and + // retained object listings are tenant-only. The control plane must not read or mutate them. switch relation { case "datasource_source_operations", "datasource_oauth_consumed_states", - "datasource_operation_attempts", "codefly_effect_receipts": + "datasource_operation_attempts", "codefly_effect_receipts", + "datasource_object_manifests": want = relationPrivileges{} } if relation == "feature_flags" { diff --git a/module/services/accounts/code/pkg/relationcatalog/catalog.go b/module/services/accounts/code/pkg/relationcatalog/catalog.go index 2151dc06d..ab2c23d56 100644 --- a/module/services/accounts/code/pkg/relationcatalog/catalog.go +++ b/module/services/accounts/code/pkg/relationcatalog/catalog.go @@ -141,6 +141,7 @@ var authorities = map[string]Authority{ "dashboards": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "org_id"}, "datasource_sources": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "org_id"}, "datasource_source_operations": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "org_id"}, + "datasource_object_manifests": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "org_id"}, "datasource_operation_attempts": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "org_id"}, "codefly_effect_receipts": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "tenant"}, "datasource_oauth_consumed_states": {Scope: ScopeTenant, PolicyShape: ShapeDirect, ScopeColumn: "org_id"}, diff --git a/module/services/store/migrations/46_datasource_object_manifests.down.sql b/module/services/store/migrations/46_datasource_object_manifests.down.sql new file mode 100644 index 000000000..07c5fa13e --- /dev/null +++ b/module/services/store/migrations/46_datasource_object_manifests.down.sql @@ -0,0 +1 @@ +DROP TABLE public.datasource_object_manifests; diff --git a/module/services/store/migrations/46_datasource_object_manifests.up.sql b/module/services/store/migrations/46_datasource_object_manifests.up.sql new file mode 100644 index 000000000..b68f79517 --- /dev/null +++ b/module/services/store/migrations/46_datasource_object_manifests.up.sql @@ -0,0 +1,23 @@ +-- The listings an object-storage source was synced at. An object store answers +-- only "what is there now", so the host keeps the listing each version names: +-- the next sync diffs the current listing against it, and a module fetches an +-- object at the version it was told of. The version is the listing's own +-- digest under the source's binding, so a row is written once and never +-- changes; only the newest few per source are kept. Tenant-owned: the control +-- plane neither reads nor writes it. +CREATE TABLE public.datasource_object_manifests ( + org_id uuid NOT NULL, + source_id uuid NOT NULL, + version text NOT NULL CHECK (version ~ '^[a-f0-9]{16}\.[a-f0-9]{64}$'), + objects jsonb NOT NULL CHECK (jsonb_typeof(objects) = 'array'), + retained_at timestamptz NOT NULL DEFAULT now(), + FOREIGN KEY (org_id,source_id) REFERENCES public.datasource_sources(org_id,id) ON DELETE CASCADE, + PRIMARY KEY (org_id,source_id,version) +); +CREATE INDEX datasource_object_manifests_retained ON public.datasource_object_manifests (org_id,source_id,retained_at DESC); +ALTER TABLE public.datasource_object_manifests ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.datasource_object_manifests FORCE ROW LEVEL SECURITY; +CREATE POLICY datasource_object_manifests_tenant ON public.datasource_object_manifests TO app_tenant + USING (org_id = NULLIF(current_setting('app.current_org_id', true), '')::uuid) + WITH CHECK (org_id = NULLIF(current_setting('app.current_org_id', true), '')::uuid); +GRANT SELECT, INSERT, UPDATE, DELETE ON public.datasource_object_manifests TO app_tenant;