From 3226a1a12bcf6018504274ea6f1043372c54082a Mon Sep 17 00:00:00 2001 From: antoinetoussaint-byte <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Tue, 27 Oct 2026 11:29:25 -0400 Subject: [PATCH] feat(accounts): the object-storage datasource provider is resolved from the composed service, not compiled into the registry The connector registry is built with the compiled-in providers only (github, api, crawler). Object storage is registered through Registry.Register, wrapped in connector.Budgeted, when the declared object-storage dependency resolves to a service that can read a bucket (SetObjectStorage). A composition without that service has no upload entry in the catalog, and AddSource and sync refuse a source of it by name. Everything the host does for the provider moves to datasource_objectstore.go; datasources.go no longer imports objectstore, and the sync batches by the registered descriptor's budget. The conformance suite also runs against the connector the registry holds. Closes #1268 Co-Authored-By: Claude Opus 5.5 --- module/services/README.md | 6 +- module/services/accounts/AGENTS.md | 54 +++- .../pkg/business/datasource_connectors.go | 94 ++---- .../business/datasource_connectors_test.go | 34 ++- ...atasource_object_storage_admission_test.go | 9 +- .../pkg/business/datasource_objectstore.go | 280 ++++++++++++++++++ ...datasource_objectstore_conformance_test.go | 174 +++++++++++ .../accounts/code/pkg/business/datasources.go | 174 +---------- .../accounts/code/pkg/business/export_test.go | 7 + .../pkg/datasource/connector/CONFORMANCE.md | 24 +- module/services/accounts/code/work.go | 12 +- 11 files changed, 584 insertions(+), 284 deletions(-) create mode 100644 module/services/accounts/code/pkg/business/datasource_objectstore.go create mode 100644 module/services/accounts/code/pkg/business/datasource_objectstore_conformance_test.go create mode 100644 module/services/accounts/code/pkg/business/export_test.go diff --git a/module/services/README.md b/module/services/README.md index d898179a5..7e525406c 100644 --- a/module/services/README.md +++ b/module/services/README.md @@ -13,8 +13,10 @@ behind `auth-gateway`. (codefly-dev/service-object-storage). The host keeps no store client of its own: accounts reads every object-storage datasource through it, under a per-call binding of the source's bucket and credential, and the gateway owns -the store protocol and the public-destination guard — see -`accounts/AGENTS.md`. +the store protocol and the public-destination guard. The object-storage +datasource provider exists only where this service is composed: accounts +registers it when the dependency resolves, and a composition that leaves it out +offers no object-storage sources — see `accounts/AGENTS.md`. Run the complete local dependency graph from `module/`: diff --git a/module/services/accounts/AGENTS.md b/module/services/accounts/AGENTS.md index b917fe49f..7d13f4108 100644 --- a/module/services/accounts/AGENTS.md +++ b/module/services/accounts/AGENTS.md @@ -1210,6 +1210,42 @@ prefix leaves the filter where it is. envelope connector applies it to the service's listing before the version is digested (#1211). +## A datasource provider is compiled in, or resolved from a composed service + +The connector registry (`pkg/business/datasource_connectors.go`) is built at +start with the providers compiled into this host — `github` (conformant), `api` +and `crawler` (registered non-conformant with their gap). **Object storage is +not one of them.** It is the first provider resolved from a composed service: +`datasource_objectstore.go` registers it only once the declared +`object-storage` dependency resolves to a service that can read a tenant's +bucket, and holds everything the host does for that provider, so the generic +catalog, admission and sync paths name no object-storage type. + +The resolution reuses what the registry already has — nothing beside it +decides what a provider is: + +- **Registration** goes through `Registry.Register`, so the descriptor is + validated and the interface checked exactly as for a compiled-in provider, + and the connector is wrapped in `connector.Budgeted`. +- **The per-credential budget stays host-side.** It meters across every source + and replica (`datasource_credential_budgets`), so it is shared state with a + tenancy; the composed service never sees it. +- **The credential is per call** (envelope clause 3). The host decrypts it from + the source on each call and puts it on that call's binding; the service + neither returns nor stores it. +- **The conformance suite runs against the provider as registered** + (`TestObjectStorageAsRegisteredPassesTheConformanceSuite`): the connector the + registry holds once the provider is resolved, on sources connected through + `AddSource`, reading their credential and retained listings through the host. +- **Absent means absent.** A composition that leaves the service out — or + `codefly run service … --exclude-dependency saas-starter/object-storage` — + has no `upload` entry in `GetDatasourceCatalog`, and a source of it is + refused by name, not as an unknown provider. + +The four provider constants (`github`, `api`, `crawler`, `upload`) stay until a +provider contract is published; converting `github`, `api` and `crawler` to +composed services is not done here. + ## A datasource source is refreshed by three things, and one of them covers every provider Ranked by latency, a source's content is refreshed by a push (near-live), by a @@ -1394,14 +1430,16 @@ keeps what is per-source state: `GetDatasourceCatalog` reports `accepts_new_sources` and `AddSource` takes a new source with its prefix and key filter. That is the whole admin path a composed consumer drives — catalog, `AddSource`, sync — with no portal step. -- **Admission needs a service that can read a bucket.** At start `work.go` - wires the composed service through `Service.SetObjectStorage`, which first - asks it, unbound, whether it reads a batch (`objectstore.CheckService`) and - refuses to wire one that does not. A composition without the service, or with one - from before the binding (0.0.18 and earlier; this module pins 0.0.19), - leaves object storage unavailable: the catalog says so, `AddSource` refuses - with `ErrObjectStorageUnavailable`, and an existing source's sync fails by - name. +- **The provider exists only where its service resolves.** Object storage is + not compiled into the registry (see the next section). At start `work.go` + resolves the declared `object-storage` dependency and hands it to + `Service.SetObjectStorage`, which first asks it, unbound, whether it reads a + batch (`objectstore.CheckService`) and only then registers the provider. A + composition without the service, or with one from before the binding (0.0.18 + and earlier; this module pins 0.0.19), has no `upload` provider: the catalog + does not list it, `AddSource` refuses with `ErrObjectStorageUnavailable`, an + existing source's sync fails with the same error, and the source reports it + as its conformance gap. - **A sync never empties a source that has files** — a host-owned invariant, whatever any service version answers. A closing snapshot with no files tombstones every entry mirrored from the source, and a service that wrongly diff --git a/module/services/accounts/code/pkg/business/datasource_connectors.go b/module/services/accounts/code/pkg/business/datasource_connectors.go index 3844e601f..6cccb17f1 100644 --- a/module/services/accounts/code/pkg/business/datasource_connectors.go +++ b/module/services/accounts/code/pkg/business/datasource_connectors.go @@ -7,21 +7,21 @@ import ( "accounts/pkg/datasource/connector" "accounts/pkg/datasource/github" - "accounts/pkg/datasource/objectstore" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) // The host's datasource connectors, on the envelope in pkg/datasource/connector. -// Two are conformant, each passing the conformance suite beside it: GitHub -// (pkg/datasource/github/connector_conformance_test.go) and object storage, -// read through the composed object-storage service -// (pkg/datasource/objectstore/connector_conformance_test.go). The generic API -// and the crawler still run on their own engines, outside the envelope; they -// are registered non-conformant with their gap, so their existing sources keep -// running and new ones are refused until each passes the suite (the owner's -// settled question 7). +// Two are conformant: GitHub, compiled in and passing the conformance suite +// beside it (pkg/datasource/github/connector_conformance_test.go), and object +// storage, which is not compiled into this registry — it is registered only +// when the composed object-storage service resolves, and passes the suite as +// registered (datasource_objectstore.go). The generic API and the crawler +// still run on their own engines, outside the envelope; they are registered +// non-conformant with their gap, so their existing sources keep running and +// new ones are refused until each passes the suite (the owner's settled +// question 7). // Why each legacy provider is not on the envelope yet, per connector. What // remains for each, clause by clause, and what it already honours, is in @@ -64,10 +64,12 @@ func (s *Service) newDatasourceConnectorRegistry() *connector.Registry { CredentialModes: []connector.CredentialMode{connector.CredentialNone}, Gap: gapCrawlerDatasource, })) - // Object storage reads through the composed object-storage service and - // passes the suite over its wire (objectstore/connector_conformance_test.go); - // metered like GitHub, per source credential. - must(r.Register(connector.Budgeted(objectstore.NewFilesConnector(s.uploadBucketForSource, objectManifests{s.store}), datasourceScheduler{s: s}))) + // Object storage is not compiled in: it is registered when its composed + // service resolves (datasource_objectstore.go), here only if that happened + // before this registry was built. + if s.newUploadBucket != nil { + must(s.registerObjectStorage(r)) + } return r } @@ -124,16 +126,15 @@ func DatasourceConformance(registry *connector.Registry, provider string) (bool, return false, "the datasource connector is not configured" } d, ok := registry.Descriptor(provider) - if !ok { + switch { + case !ok && provider == DatasourceProviderUpload: + return false, ErrObjectStorageUnavailable.Error() + case !ok: return false, "no connector is registered for this provider" } return d.Conformant, d.Gap } -// ErrObjectStorageUnavailable: no composed object-storage service here can read -// a tenant's bucket, so an object-storage source is neither admitted nor synced. -var ErrObjectStorageUnavailable = errors.New("object storage is not available here: no composed object-storage service serves a per-call binding with a batched read") - // admitNewDatasource refuses a new source of a provider the registry does not // admit, or that nothing composed here can serve, before anything is // validated, sealed or stored. The catalog reports the same answer. @@ -143,14 +144,13 @@ func (s *Service) admitNewDatasource(provider string) error { } err := s.datasourceConnectors.AdmitNewSource(provider) switch { - case err == nil && provider == DatasourceProviderUpload && s.newUploadBucket == nil: - // Conformant, but nothing here can read a bucket: no object-storage - // service is composed, or the one composed predates the per-call - // binding and the batched read (work.go asks it at start). A source - // admitted now would fail every sync. - return status.Error(codes.FailedPrecondition, ErrObjectStorageUnavailable.Error()) case err == nil: return nil + case errors.Is(err, connector.ErrUnknownConnector) && provider == DatasourceProviderUpload: + // Not registered: no object-storage service is composed, or the one + // composed predates the per-call binding and the batched read + // (SetObjectStorage). The refusal names why, not "unknown". + return status.Error(codes.FailedPrecondition, ErrObjectStorageUnavailable.Error()) case errors.Is(err, connector.ErrUnknownConnector): return status.Errorf(codes.InvalidArgument, "unknown datasource provider %q", provider) default: @@ -166,55 +166,11 @@ func connectorSource(source *DatasourceSource) connector.Source { case source.Provider == DatasourceProviderGitHub: src.Config = github.SourceConfig{Repo: source.Repo, Branch: source.Branch, InScope: datasourceFileScope(source)} case source.Provider == DatasourceProviderUpload && source.Upload != nil: - cfg := source.Upload - src.Config = objectstore.SourceConfig{ - Endpoint: cfg.Endpoint, Region: cfg.Region, Bucket: cfg.Bucket, Prefix: cfg.Prefix, - Scope: objectstore.Scope{InScope: uploadKeyScope(cfg), MaxObjects: cfg.MaxObjects}, - } + src.Config = uploadSourceConfig(source.Upload) } return src } -// uploadBucketForSource resolves one object-storage source's Bucket, per call: -// the host decrypts the source's secret access key and binds it to that call, -// and the connector never sees it. -func (s *Service) uploadBucketForSource(ctx context.Context, src connector.Source) (objectstore.Bucket, error) { - if s.newUploadBucket == nil { - return nil, ErrObjectStorageUnavailable - } - source, err := s.store.GetDatasourceSourceByID(ctx, src.ID) - if err != nil { - return nil, err - } - if source == nil || source.OrgID != src.OrgID || source.Provider != DatasourceProviderUpload || source.Upload == nil { - return nil, fmt.Errorf("datasource source %s is not an object-storage source of org %s", src.ID, src.OrgID) - } - secret, err := s.datasourceCipher.DecryptSecret(ctx, DatasourceConnectorSecretPurpose(source.ID), source.CredentialSecretRef) - if err != nil { - return nil, fmt.Errorf("decrypt secret access key: %w", err) - } - cfg, _ := connectorSource(source).Config.(objectstore.SourceConfig) - return s.newUploadBucket(cfg, source.Upload.AccessKeyID, secret), nil -} - -// objectManifests retains an object-storage source's listings in the store, -// inside the source's org transaction. -type objectManifests struct{ store Store } - -func (m objectManifests) Put(ctx context.Context, src connector.Source, version string, items []objectstore.Item) error { - return m.store.WithOrgTx(ctx, src.OrgID, func(ctx context.Context) error { - return m.store.PutDatasourceObjectManifest(ctx, src.OrgID, src.ID, version, items) - }) -} - -func (m objectManifests) Get(ctx context.Context, src connector.Source, version string) (items []objectstore.Item, ok bool, err error) { - err = m.store.WithOrgTx(ctx, src.OrgID, func(ctx context.Context) error { - items, ok, err = m.store.GetDatasourceObjectManifest(ctx, src.OrgID, src.ID, version) - return err - }) - return items, ok, err -} - // githubRemoteForSource resolves the credentialed GitHub client for one source, // per call: the host holds the credential and the connector never sees it. func (s *Service) githubRemoteForSource(ctx context.Context, src connector.Source) (github.Remote, error) { diff --git a/module/services/accounts/code/pkg/business/datasource_connectors_test.go b/module/services/accounts/code/pkg/business/datasource_connectors_test.go index 269ba244b..02503bd59 100644 --- a/module/services/accounts/code/pkg/business/datasource_connectors_test.go +++ b/module/services/accounts/code/pkg/business/datasource_connectors_test.go @@ -30,6 +30,8 @@ var suiteCall = regexp.MustCompile(`connectortest\.Run[A-Z][A-Za-z]*\(t,`) func TestEveryConformantConnectorRunsTheSuite(t *testing.T) { svc, _ := business.NewService(nil) svc.SetDatasourceConnector(nil, nil, "") + // Object storage is registered once its composed service resolves. + svc.SetDatasourceUploadBucketFactory(func(objectstore.SourceConfig, string, string) objectstore.Bucket { return nil }) conformant := svc.DatasourceConnectors().Conformant() if len(conformant) < 2 { t.Fatalf("the host registers %d conformant connectors, want GitHub and object storage", len(conformant)) @@ -108,25 +110,26 @@ func TestLegacyProvidersAreRegisteredWithTheirGap(t *testing.T) { } } -// TestObjectStorageIsAdmittedOnlyWhereAServiceCanReadIt: object storage is -// conformant, but a host with no composed object-storage service — or one -// whose service predates the per-call binding, which work.go leaves unwired — -// cannot sync a bucket. The catalog says so, and AddSource refuses by name -// before anything is sealed or stored. -func TestObjectStorageIsAdmittedOnlyWhereAServiceCanReadIt(t *testing.T) { +// TestObjectStorageIsRegisteredOnlyWhereItsServiceResolves: object storage is +// not compiled into the registry. A host with no composed object-storage +// service — or one whose service predates the per-call binding, which +// SetObjectStorage refuses — has no such provider: the catalog does not list +// it, AddSource refuses a source of it by name before anything is sealed or +// stored, and an existing source reports why. Resolving the service registers +// it, conformant and admitted. +func TestObjectStorageIsRegisteredOnlyWhereItsServiceResolves(t *testing.T) { svc, _ := business.NewService(nil) svc.SetDatasourceConnector(nil, nil, "") - accepts := func() bool { + listed := func() (present, accepts bool) { for _, e := range svc.DatasourceCatalog() { if e.Descriptor.Key == business.DatasourceProviderUpload { - return e.AcceptsNewSources + return true, e.AcceptsNewSources } } - t.Fatal("object storage is not in the catalog") - return false + return false, false } - if accepts() { - t.Fatal("the catalog accepts object-storage sources with no object-storage service composed") + if present, _ := listed(); present { + t.Fatal("the catalog lists object storage with no object-storage service composed") } _, err := svc.AddSource(context.Background(), "actor-1", business.AddSourceInput{ OrgID: "11111111-1111-1111-1111-111111111111", Provider: business.DatasourceProviderUpload, CollectionLabel: "guides", Credential: "secretkey", @@ -135,8 +138,11 @@ func TestObjectStorageIsAdmittedOnlyWhereAServiceCanReadIt(t *testing.T) { if status.Code(err) != codes.FailedPrecondition || !strings.Contains(err.Error(), business.ErrObjectStorageUnavailable.Error()) { t.Fatalf("AddSource(upload) with no object storage = %v, want the unavailable refusal", err) } + if conformant, gap := business.DatasourceConformance(svc.DatasourceConnectors(), business.DatasourceProviderUpload); conformant || gap != business.ErrObjectStorageUnavailable.Error() { + t.Fatalf("an existing object-storage source with no service reports (%v, %q), want the unavailable reason", conformant, gap) + } svc.SetDatasourceUploadBucketFactory(func(objectstore.SourceConfig, string, string) objectstore.Bucket { return nil }) - if !accepts() { - t.Fatal("the catalog refuses object-storage sources with an object-storage service composed") + if present, accepts := listed(); !present || !accepts { + t.Fatalf("with an object-storage service composed the catalog lists object storage = %v, accepting = %v", present, accepts) } } diff --git a/module/services/accounts/code/pkg/business/datasource_object_storage_admission_test.go b/module/services/accounts/code/pkg/business/datasource_object_storage_admission_test.go index 1556ee877..8797623b6 100644 --- a/module/services/accounts/code/pkg/business/datasource_object_storage_admission_test.go +++ b/module/services/accounts/code/pkg/business/datasource_object_storage_admission_test.go @@ -52,10 +52,10 @@ func composedStorageClient(t *testing.T, binding bool) storagev0.ObjectStorageCl // review r1's major 3, pinned where the defect was: with the released 0.0.18 // composed — a service that answers, but cannot serve a binding — the catalog // accepted new object-storage sources and AddSource stored them, and only each -// sync then failed. Wiring that service is refused, and the refusal is at -// admission: the catalog does not offer the provider, AddSource refuses by -// name, and nothing is stored or audited. A service that serves the binding is -// admitted. +// sync then failed. Wiring that service is refused and registers nothing, so +// the refusal is at admission: the catalog does not list the provider, +// AddSource refuses by name, and nothing is stored or audited. A service that +// serves the binding registers it. func TestObjectStorageIsRefusedAtAdmissionWhenTheComposedServiceCannotServeIt(t *testing.T) { ctx := context.Background() accepts := func(svc *business.Service) bool { @@ -64,7 +64,6 @@ func TestObjectStorageIsRefusedAtAdmissionWhenTheComposedServiceCannotServeIt(t return e.AcceptsNewSources } } - t.Fatal("object storage is not in the catalog") return false } input := business.AddSourceInput{ diff --git a/module/services/accounts/code/pkg/business/datasource_objectstore.go b/module/services/accounts/code/pkg/business/datasource_objectstore.go new file mode 100644 index 000000000..f9faf5f4d --- /dev/null +++ b/module/services/accounts/code/pkg/business/datasource_objectstore.go @@ -0,0 +1,280 @@ +package business + +import ( + "context" + "errors" + "fmt" + "io" + "time" + + "accounts/pkg/datasource/connector" + "accounts/pkg/datasource/objectstore" + + "github.com/codefly-dev/core/wool" + storagev0 "github.com/codefly-dev/service-object-storage/gen/codefly/storage/v0" +) + +// Object storage is the first datasource provider the host resolves from a +// composed service instead of compiling it into its registry. The registry +// built at start (newDatasourceConnectorRegistry) holds the compiled-in +// providers only; object storage is registered when, and only when, the +// declared object-storage dependency resolves to a composed service that can +// read a tenant's bucket (SetObjectStorage, called from work.go). A +// composition without that service has no object-storage provider at all: the +// catalog does not list it, AddSource refuses a source of it by name, and an +// existing source's sync refuses with the same error. Everything the host does +// for the provider is in this file; the generic sync, catalog and admission +// paths name no object-storage type. + +// ErrObjectStorageUnavailable: no composed object-storage service here can read +// a tenant's bucket, so the provider is not registered and an object-storage +// source is neither admitted nor synced. +var ErrObjectStorageUnavailable = errors.New("object storage is not available here: no composed object-storage service serves a per-call binding with a batched read") + +// ErrUploadDestinationRefused: the object-storage service's guard refused a +// source's endpoint. The source is parked; its endpoint must change. +var ErrUploadDestinationRefused = objectstore.ErrDestinationRefused + +// uploadBucketFactory builds one object-storage source's Bucket for one call. +type uploadBucketFactory = func(cfg objectstore.SourceConfig, accessKeyID, secretAccessKey string) objectstore.Bucket + +// SetObjectStorage resolves the object-storage provider from the composed +// object-storage service: every object-storage source is read through it, under +// a per-call binding of the source's bucket and credential — if, and only if, +// the service can serve one: it asks first (objectstore.CheckService). A +// service that cannot — a release from before the binding, such as 0.0.18 — is +// refused and the provider is not registered, exactly as when none is composed. +func (s *Service) SetObjectStorage(ctx context.Context, client storagev0.ObjectStorageClient) error { + if err := objectstore.CheckService(ctx, client); err != nil { + return err + } + return s.useObjectStorage(func(cfg objectstore.SourceConfig, accessKeyID, secretAccessKey string) objectstore.Bucket { + return objectstore.NewServiceBucket(client, cfg, accessKeyID, secretAccessKey) + }) +} + +// SetDatasourceUploadBucketFactory registers the object-storage provider over +// factory in place of the service. Tests use it to script a store. +func (s *Service) SetDatasourceUploadBucketFactory(factory uploadBucketFactory) { + if err := s.useObjectStorage(factory); err != nil { + panic(fmt.Sprintf("datasource connector registry: %v", err)) + } +} + +// useObjectStorage makes factory the bucket every object-storage source reads +// and registers the provider in the host's registry, once. A registry built +// later registers it from the factory (newDatasourceConnectorRegistry). +func (s *Service) useObjectStorage(factory uploadBucketFactory) error { + s.newUploadBucket = factory + if s.datasourceConnectors == nil { + return nil + } + return s.registerObjectStorage(s.datasourceConnectors) +} + +// registerObjectStorage adds the object-storage provider to r through the +// registry's own validation, metered per source credential like GitHub. It +// passes the conformance suite as registered, over the service's wire +// (TestObjectStorageAsRegisteredPassesTheConformanceSuite). +func (s *Service) registerObjectStorage(r *connector.Registry) error { + if _, registered := r.Descriptor(DatasourceProviderUpload); registered { + return nil + } + return r.Register(connector.Budgeted(objectstore.NewFilesConnector(s.uploadBucketForSource, objectManifests{s.store}), datasourceScheduler{s: s})) +} + +// uploadSourceConfig is the connector's view of a stored object-storage +// source's configuration. +func uploadSourceConfig(cfg *UploadDatasourceConfig) objectstore.SourceConfig { + return objectstore.SourceConfig{ + Endpoint: cfg.Endpoint, Region: cfg.Region, Bucket: cfg.Bucket, Prefix: cfg.Prefix, + Scope: objectstore.Scope{InScope: uploadKeyScope(cfg), MaxObjects: cfg.MaxObjects}, + } +} + +// uploadBucketForSource resolves one object-storage source's Bucket, per call: +// the host decrypts the source's secret access key and binds it to that call, +// and the connector never sees it. +func (s *Service) uploadBucketForSource(ctx context.Context, src connector.Source) (objectstore.Bucket, error) { + if s.newUploadBucket == nil { + return nil, ErrObjectStorageUnavailable + } + source, err := s.store.GetDatasourceSourceByID(ctx, src.ID) + if err != nil { + return nil, err + } + if source == nil || source.OrgID != src.OrgID || source.Provider != DatasourceProviderUpload || source.Upload == nil { + return nil, fmt.Errorf("datasource source %s is not an object-storage source of org %s", src.ID, src.OrgID) + } + secret, err := s.datasourceCipher.DecryptSecret(ctx, DatasourceConnectorSecretPurpose(source.ID), source.CredentialSecretRef) + if err != nil { + return nil, fmt.Errorf("decrypt secret access key: %w", err) + } + return s.newUploadBucket(uploadSourceConfig(source.Upload), source.Upload.AccessKeyID, secret), nil +} + +// objectManifests retains an object-storage source's listings in the store, +// inside the source's org transaction. +type objectManifests struct{ store Store } + +func (m objectManifests) Put(ctx context.Context, src connector.Source, version string, items []objectstore.Item) error { + return m.store.WithOrgTx(ctx, src.OrgID, func(ctx context.Context) error { + return m.store.PutDatasourceObjectManifest(ctx, src.OrgID, src.ID, version, items) + }) +} + +func (m objectManifests) Get(ctx context.Context, src connector.Source, version string) (items []objectstore.Item, ok bool, err error) { + err = m.store.WithOrgTx(ctx, src.OrgID, func(ctx context.Context) error { + items, ok, err = m.store.GetDatasourceObjectManifest(ctx, src.OrgID, src.ID, version) + return err + }) + return items, ok, err +} + +// runUploadSync mirrors the source's scope into the documents store. It takes a +// complete snapshot through the object-storage connector — the listing within +// the source's key filter and object cap, read through the object-storage +// service — and fetches it in batches, enqueuing an ingest delivery per +// object, then closes the sync with the complete listing of every object it +// delivered, so the store tombstones the entries whose objects are gone. Every +// job of one sync carries one freshly allocated ordinal, which the store ranks +// the listing against the writes it covers by. +// +// The listing is sent only when the sync saw the whole scope and delivered +// every object in it: a failed listing, one the cap cuts, a fetch failure, or +// an object too large to deliver ends the sync without one, so a partial view +// can never delete. An object rewritten or deleted between the listing and its +// read fails the sync, which the next one lists afresh. +func (s *Service) runUploadSync(ctx context.Context, source *DatasourceSource) (int, error) { + w := wool.Get(ctx).In("runUploadSync") + // The registered connector is the metered one: every listing and batched + // read spends the source credential's budget, as a GitHub sync does. + conn, ok := s.datasourceConnectors.Files(DatasourceProviderUpload) + if !ok { + return 0, w.Wrapf(ErrObjectStorageUnavailable, "sync %s", source.ID) + } + budget := conn.Descriptor().Budget + if source.Upload == nil { + return 0, w.NewError("upload source has no config") + } + src := connectorSource(source) + snapshot, err := conn.Changes(ctx, src, "") + if errors.Is(err, objectstore.ErrListingIncomplete) { + return 0, w.Wrapf(fmt.Errorf("%w: %w", ErrUploadListingIncomplete, err), "list objects under %q", source.Upload.Prefix) + } + if err != nil { + return 0, w.Wrapf(s.parkRefusedUploadSource(ctx, source, err), "list objects") + } + // A host-owned invariant, whatever any service version answers: a sync + // never closes with a snapshot that empties a source that has files. Such a + // snapshot tombstones every entry mirrored from the source, and a store or + // service that wrongly answers "empty" — a rewritten scope, a discarded + // truncation — is the case that has already happened. Emptying a source is + // an explicit operation, not a side effect of a listing. + if len(snapshot.Changes) == 0 { + held, err := s.uploadSourceMayHoldFiles(ctx, source) + if err != nil { + return 0, w.Wrapf(err, "read the source's last snapshot") + } + if held { + return 0, w.Wrapf(ErrUploadEmptyListingRefused, "listing under %q", source.Upload.Prefix) + } + } + ordinal, err := s.allocateOrdinal(ctx, source.ID) + if err != nil { + return 0, w.Wrapf(err, "allocate ordinal") + } + + enqueued, oversized := 0, 0 + delivered := make([]uploadListingFile, 0, len(snapshot.Changes)) + var batch []connector.FileRef + var batchBytes int64 + flush := func() error { + if len(batch) == 0 { + return nil + } + err := conn.FetchFiles(ctx, src, snapshot.To, batch, func(f connector.File, content io.Reader) error { + body, err := io.ReadAll(content) + if err != nil { + return err + } + fingerprint := uploadFingerprint(f.Provenance.GetItemVersion(), body) + if err := s.enqueueUploadIngest(ctx, source, f.Locator, body, f.ContentType, fingerprint, ordinal); err != nil { + return w.Wrapf(err, "enqueue %s", f.Locator) + } + delivered = append(delivered, uploadListingFile{Key: f.Locator, ETag: fingerprint}) + enqueued++ + return nil + }) + batch, batchBytes = batch[:0], 0 + // An object the object-storage service refuses as over its cap is + // skipped and the rest of the batch is delivered; like an object over + // the ingest limit, it keeps this sync from sending a listing that + // would tombstone it. + var skipped *objectstore.TooLargeError + if errors.As(err, &skipped) { + oversized += len(skipped.Keys) + return nil + } + if err != nil { + return s.parkRefusedUploadSource(ctx, source, err) + } + return nil + } + for _, ch := range snapshot.Changes { + // Objects too large for the ingest inbox cannot be delivered and the + // consumer holds no credentials to re-fetch them: never read, counted. + if ch.Size > maxIngestPayload { + oversized++ + continue + } + if len(batch) == budget.MaxItemsPerCall || batchBytes+ch.Size > budget.MaxBytesPerCall { + if err := flush(); err != nil { + return enqueued, w.Wrapf(err, "fetch objects") + } + } + batch = append(batch, connector.FileRef{ItemID: ch.Locator, ItemVersion: ch.ItemVersion}) + batchBytes += ch.Size + } + if err := flush(); err != nil { + return enqueued, w.Wrapf(err, "fetch objects") + } + + if oversized > 0 { + return enqueued, w.NewError("%d object(s) exceed the ingest payload limit", oversized) + } + if err := s.enqueueUploadListing(ctx, source, ordinal, delivered); err != nil { + return enqueued, w.Wrapf(err, "enqueue listing") + } + + if err := s.store.WithOrgTx(ctx, source.OrgID, func(ctx context.Context) error { + return s.store.RecordDatasourceSnapshotSent(ctx, source.OrgID, source.ID, time.Now().UTC(), len(delivered)) + }); err != nil { + return enqueued, w.Wrapf(err, "record sync time") + } + return enqueued, nil +} + +// parkRefusedUploadSource parks an object-storage source a retry cannot fix: +// the object-storage service's guard refused its endpoint, or its store +// rejected its access key. The source is degraded with a reason a person can +// act on — change the endpoint, or connect it again with a valid key — so the +// reconcile sweep stops retrying it. Any other error is returned as it is. A +// park that fails returns its own error in place of the refusal, so the job +// retries and the park is attempted again (parkDatasourceSource). +func (s *Service) parkRefusedUploadSource(ctx context.Context, source *DatasourceSource, err error) error { + var reason DatasourceDegradeReason + switch { + case errors.Is(err, objectstore.ErrDestinationRefused): + reason = ObjectStorageEndpointRefusedDegradeReason() + case errors.Is(err, connector.ErrCredentialRefused): + reason = ObjectStorageCredentialRejectedDegradeReason() + default: + return err + } + if _, perr := s.parkDatasourceSource(ctx, source, reason); perr != nil { + wool.Get(ctx).Warn("could not park a refused object-storage source", wool.Field("source_id", source.ID), wool.ErrField(perr)) + return fmt.Errorf("park the refused object-storage source: %w", perr) + } + return err +} diff --git a/module/services/accounts/code/pkg/business/datasource_objectstore_conformance_test.go b/module/services/accounts/code/pkg/business/datasource_objectstore_conformance_test.go new file mode 100644 index 000000000..c5b033965 --- /dev/null +++ b/module/services/accounts/code/pkg/business/datasource_objectstore_conformance_test.go @@ -0,0 +1,174 @@ +package business_test + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "accounts/pkg/business" + "accounts/pkg/datasource/connector" + "accounts/pkg/datasource/connector/connectortest" + "accounts/pkg/datasource/objectstore" +) + +// TestObjectStorageAsRegisteredPassesTheConformanceSuite runs the suite against +// the provider the host's registry holds once object storage is resolved — +// not against the objectstore package (whose own run is over the service's +// wire): the registry's validation, the budget meter, the per-call credential +// the host decrypts from the stored source, and the listings retained in the +// host's store, on sources connected through AddSource. +func TestObjectStorageAsRegisteredPassesTheConformanceSuite(t *testing.T) { + connectortest.RunFiles(t, newRegisteredObjectStorage) +} + +const registeredStorageSecret = "sentinel-secret-key-77e2" + +type registeredObjectStorage struct { + svc *business.Service + store *datasourceFakeStore + conn connector.FilesConnector + + mu sync.Mutex + objects map[string]string + calls atomic.Int64 + throttled atomic.Bool + presented atomic.Bool +} + +func newRegisteredObjectStorage(t *testing.T) connectortest.FilesFixture { + f := ®isteredObjectStorage{store: newDatasourceFakeStore(), objects: map[string]string{}} + f.svc, _ = newDatasourceService(f.store, &recordingProducer{}, nil) + if _, ok := f.svc.DatasourceConnectors().Files(business.DatasourceProviderUpload); ok { + t.Fatal("object storage is registered before its service resolved") + } + f.svc.SetDatasourceUploadBucketFactory(func(_ objectstore.SourceConfig, _, secret string) objectstore.Bucket { + return registeredBucket{f: f, secret: secret} + }) + conn, ok := f.svc.DatasourceConnectors().Files(business.DatasourceProviderUpload) + if !ok { + t.Fatal("object storage is not registered once its service resolved") + } + f.conn = conn + return f +} + +func (f *registeredObjectStorage) Connector() connector.FilesConnector { return f.conn } + +func (f *registeredObjectStorage) Source(t *testing.T, org string) connector.Source { + source, err := f.svc.AddSource(context.Background(), "actor-1", business.AddSourceInput{ + OrgID: org, Provider: business.DatasourceProviderUpload, CollectionLabel: fmt.Sprintf("guides-%d", f.calls.Load()), + Credential: registeredStorageSecret, + // The suite lists 120 objects: the source takes the whole bucket at + // the object cap's ceiling. + Upload: &business.UploadDatasourceConfig{Endpoint: "https://store.example.com", Region: "us-east-1", Bucket: "acme-docs", AccessKeyID: "AKIAEXAMPLE", MaxObjects: 1000}, + }) + if err != nil { + t.Fatal(err) + } + forgetConnectSync(f.svc, source.ID) + return business.ConnectorSource(source) +} + +func (f *registeredObjectStorage) Apply(t *testing.T, puts map[string]string, deletes []string, moves map[string]string) { + f.mu.Lock() + defer f.mu.Unlock() + for key, content := range puts { + f.objects[key] = content + } + for _, key := range deletes { + delete(f.objects, key) + } + for from, to := range moves { + f.objects[to] = f.objects[from] + delete(f.objects, from) + } +} + +// Rewrite: a bucket has no history; what makes a version undiffable is that +// the host no longer retains its listing. +func (f *registeredObjectStorage) Rewrite(t *testing.T) { + f.store.mu.Lock() + defer f.store.mu.Unlock() + clear(f.store.manifests) +} + +func (f *registeredObjectStorage) ProviderCalls() int64 { return f.calls.Load() } + +func (f *registeredObjectStorage) RateLimit(t *testing.T) func() { + f.throttled.Store(true) + return func() { f.throttled.Store(false) } +} + +func (f *registeredObjectStorage) Credential() string { return registeredStorageSecret } +func (f *registeredObjectStorage) CredentialPresented() bool { return f.presented.Load() } + +func (f *registeredObjectStorage) MaxItemBytes() int64 { + return f.conn.Descriptor().Budget.MaxItemBytes +} + +// registeredBucket is the store under one call's binding: it serves only the +// secret the host decrypted for that call. +type registeredBucket struct { + f *registeredObjectStorage + secret string +} + +func (b registeredBucket) call(ctx context.Context) error { + b.f.calls.Add(1) + if err := ctx.Err(); err != nil { + return err + } + if b.secret != registeredStorageSecret { + return connector.ErrCredentialRefused + } + b.f.presented.Store(true) + if b.f.throttled.Load() { + return &connector.RateLimitedError{ResetAt: time.Now().Add(time.Minute), Scope: "credential", Estimated: true} + } + return nil +} + +func registeredETag(content string) string { + sum := sha256.Sum256([]byte(content)) + return hex.EncodeToString(sum[:8]) +} + +func (b registeredBucket) List(ctx context.Context) ([]objectstore.Item, error) { + if err := b.call(ctx); err != nil { + return nil, err + } + b.f.mu.Lock() + defer b.f.mu.Unlock() + items := make([]objectstore.Item, 0, len(b.f.objects)) + for key, content := range b.f.objects { + items = append(items, objectstore.Item{Key: key, ETag: registeredETag(content), Size: int64(len(content))}) + } + sort.Slice(items, func(i, j int) bool { return items[i].Key < items[j].Key }) + return items, nil +} + +func (b registeredBucket) Read(ctx context.Context, items []objectstore.Item, visit func(objectstore.Item, string, io.Reader) error) error { + if err := b.call(ctx); err != nil { + return err + } + for _, it := range items { + b.f.mu.Lock() + content, ok := b.f.objects[it.Key] + b.f.mu.Unlock() + if !ok || registeredETag(content) != it.ETag { + return objectstore.ErrObjectChanged + } + if err := visit(it, "text/markdown", strings.NewReader(content)); err != nil { + return err + } + } + return nil +} diff --git a/module/services/accounts/code/pkg/business/datasources.go b/module/services/accounts/code/pkg/business/datasources.go index 92c2fb7d3..fc5a7ca74 100644 --- a/module/services/accounts/code/pkg/business/datasources.go +++ b/module/services/accounts/code/pkg/business/datasources.go @@ -17,13 +17,11 @@ import ( "accounts/pkg/datasource/connector" "accounts/pkg/datasource/crawler" "accounts/pkg/datasource/github" - "accounts/pkg/datasource/objectstore" gen "accounts/pkg/gen/saas/accounts/v1" jobsv1 "accounts/pkg/gen/saas/jobs/v1" "accounts/pkg/jobs" "github.com/codefly-dev/core/wool" - storagev0 "github.com/codefly-dev/service-object-storage/gen/codefly/storage/v0" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -616,29 +614,6 @@ func (s *Service) SetDatasourceCrawlerClientFactory(factory func(cfg CrawlerData s.newCrawlerClient = factory } -// SetObjectStorage makes the composed object-storage service the store every -// object-storage source is read through, under a per-call binding of the -// source's bucket and credential — if, and only if, the service can serve one: -// it asks first (objectstore.CheckService). A service that cannot — a release -// from before the binding, such as 0.0.18 — is refused and left unwired, so -// object storage is unavailable: the catalog does not offer it, AddSource -// refuses it, and nothing is admitted that every sync would then fail on. -func (s *Service) SetObjectStorage(ctx context.Context, client storagev0.ObjectStorageClient) error { - if err := objectstore.CheckService(ctx, client); err != nil { - return err - } - s.newUploadBucket = func(cfg objectstore.SourceConfig, accessKeyID, secretAccessKey string) objectstore.Bucket { - return objectstore.NewServiceBucket(client, cfg, accessKeyID, secretAccessKey) - } - return nil -} - -// SetDatasourceUploadBucketFactory overrides how an object-storage source's -// Bucket is built. Tests use it to script a store without the service. -func (s *Service) SetDatasourceUploadBucketFactory(factory func(cfg objectstore.SourceConfig, accessKeyID, secretAccessKey string) objectstore.Bucket) { - s.newUploadBucket = factory -} - // SetDatasourceOAuth2RefreshFunc overrides how an OAuth 2.0 access token is // refreshed. Tests use it to inject a deterministic token without a live token // endpoint. @@ -1933,129 +1908,6 @@ func crawlerIngestIdempotencyKey(sourceID, pageURL, contentSHA string) string { // fails the sync terminally rather than retrying into the same cap. var ErrUploadListingIncomplete = errors.New("upload listing is incomplete; removals were not reconciled") -// runUploadSync mirrors the source's scope into the documents store. It takes a -// complete snapshot through the object-storage connector — the listing within -// the source's key filter and object cap, read through the object-storage -// service — and fetches it in batches, enqueuing an ingest delivery per -// object, then closes the sync with the complete listing of every object it -// delivered, so the store tombstones the entries whose objects are gone. Every -// job of one sync carries one freshly allocated ordinal, which the store ranks -// the listing against the writes it covers by. -// -// The listing is sent only when the sync saw the whole scope and delivered -// every object in it: a failed listing, one the cap cuts, a fetch failure, or -// an object too large to deliver ends the sync without one, so a partial view -// can never delete. An object rewritten or deleted between the listing and its -// read fails the sync, which the next one lists afresh. -func (s *Service) runUploadSync(ctx context.Context, source *DatasourceSource) (int, error) { - w := wool.Get(ctx).In("runUploadSync") - // The registered connector is the metered one: every listing and batched - // read spends the source credential's budget, as a GitHub sync does. - conn, ok := s.datasourceConnectors.Files(DatasourceProviderUpload) - if !ok { - return 0, w.NewError("datasource connector is not configured") - } - if source.Upload == nil { - return 0, w.NewError("upload source has no config") - } - src := connectorSource(source) - snapshot, err := conn.Changes(ctx, src, "") - if errors.Is(err, objectstore.ErrListingIncomplete) { - return 0, w.Wrapf(fmt.Errorf("%w: %w", ErrUploadListingIncomplete, err), "list objects under %q", source.Upload.Prefix) - } - if err != nil { - return 0, w.Wrapf(s.parkRefusedUploadSource(ctx, source, err), "list objects") - } - // A host-owned invariant, whatever any service version answers: a sync - // never closes with a snapshot that empties a source that has files. Such a - // snapshot tombstones every entry mirrored from the source, and a store or - // service that wrongly answers "empty" — a rewritten scope, a discarded - // truncation — is the case that has already happened. Emptying a source is - // an explicit operation, not a side effect of a listing. - if len(snapshot.Changes) == 0 { - held, err := s.uploadSourceMayHoldFiles(ctx, source) - if err != nil { - return 0, w.Wrapf(err, "read the source's last snapshot") - } - if held { - return 0, w.Wrapf(ErrUploadEmptyListingRefused, "listing under %q", source.Upload.Prefix) - } - } - ordinal, err := s.allocateOrdinal(ctx, source.ID) - if err != nil { - return 0, w.Wrapf(err, "allocate ordinal") - } - - enqueued, oversized := 0, 0 - delivered := make([]uploadListingFile, 0, len(snapshot.Changes)) - var batch []connector.FileRef - var batchBytes int64 - flush := func() error { - if len(batch) == 0 { - return nil - } - err := conn.FetchFiles(ctx, src, snapshot.To, batch, func(f connector.File, content io.Reader) error { - body, err := io.ReadAll(content) - if err != nil { - return err - } - fingerprint := uploadFingerprint(f.Provenance.GetItemVersion(), body) - if err := s.enqueueUploadIngest(ctx, source, f.Locator, body, f.ContentType, fingerprint, ordinal); err != nil { - return w.Wrapf(err, "enqueue %s", f.Locator) - } - delivered = append(delivered, uploadListingFile{Key: f.Locator, ETag: fingerprint}) - enqueued++ - return nil - }) - batch, batchBytes = batch[:0], 0 - // An object the object-storage service refuses as over its cap is - // skipped and the rest of the batch is delivered; like an object over - // the ingest limit, it keeps this sync from sending a listing that - // would tombstone it. - var skipped *objectstore.TooLargeError - if errors.As(err, &skipped) { - oversized += len(skipped.Keys) - return nil - } - if err != nil { - return s.parkRefusedUploadSource(ctx, source, err) - } - return nil - } - for _, ch := range snapshot.Changes { - // Objects too large for the ingest inbox cannot be delivered and the - // consumer holds no credentials to re-fetch them: never read, counted. - if ch.Size > maxIngestPayload { - oversized++ - continue - } - if len(batch) == objectstore.MaxObjectsPerBatch || batchBytes+ch.Size > objectstore.MaxBatchBytes { - if err := flush(); err != nil { - return enqueued, w.Wrapf(err, "fetch objects") - } - } - batch = append(batch, connector.FileRef{ItemID: ch.Locator, ItemVersion: ch.ItemVersion}) - batchBytes += ch.Size - } - if err := flush(); err != nil { - return enqueued, w.Wrapf(err, "fetch objects") - } - - if oversized > 0 { - return enqueued, w.NewError("%d object(s) exceed the ingest payload limit", oversized) - } - if err := s.enqueueUploadListing(ctx, source, ordinal, delivered); err != nil { - return enqueued, w.Wrapf(err, "enqueue listing") - } - - if err := s.store.WithOrgTx(ctx, source.OrgID, func(ctx context.Context) error { - return s.store.RecordDatasourceSnapshotSent(ctx, source.OrgID, source.ID, time.Now().UTC(), len(delivered)) - }); err != nil { - return enqueued, w.Wrapf(err, "record sync time") - } - return enqueued, nil -} - // ErrUploadEmptyListingRefused reports an object-storage sync whose listing // came back empty for a source that has files. It is refused, not closed: the // collection is left as it is, and emptying a source is an explicit operation. @@ -2082,30 +1934,6 @@ func (s *Service) uploadSourceMayHoldFiles(ctx context.Context, source *Datasour return source.LastSyncedAt != nil, nil } -// parkRefusedUploadSource parks an object-storage source a retry cannot fix: -// the object-storage service's guard refused its endpoint, or its store -// rejected its access key. The source is degraded with a reason a person can -// act on — change the endpoint, or connect it again with a valid key — so the -// reconcile sweep stops retrying it. Any other error is returned as it is. A -// park that fails returns its own error in place of the refusal, so the job -// retries and the park is attempted again (parkDatasourceSource). -func (s *Service) parkRefusedUploadSource(ctx context.Context, source *DatasourceSource, err error) error { - var reason DatasourceDegradeReason - switch { - case errors.Is(err, objectstore.ErrDestinationRefused): - reason = ObjectStorageEndpointRefusedDegradeReason() - case errors.Is(err, connector.ErrCredentialRefused): - reason = ObjectStorageCredentialRejectedDegradeReason() - default: - return err - } - if _, perr := s.parkDatasourceSource(ctx, source, reason); perr != nil { - wool.Get(ctx).Warn("could not park a refused object-storage source", wool.Field("source_id", source.ID), wool.ErrField(perr)) - return fmt.Errorf("park the refused object-storage source: %w", perr) - } - return err -} - // uploadListingFile is one delivered object in an upload sync's listing: its // key and the fingerprint its own ingest job was versioned by. type uploadListingFile struct { @@ -2247,7 +2075,7 @@ func (s *Service) NewDatasourceSyncJobHandler() jobs.Handler { case errors.Is(err, ErrUploadEmptyListingRefused): // The collection is left as it is; a retry would list the same. return jobs.NewProcessingError("datasource.upload_empty_listing_refused", ErrUploadEmptyListingRefused.Error(), false) - case errors.Is(err, objectstore.ErrDestinationRefused): + case errors.Is(err, ErrUploadDestinationRefused): // The source is parked: its endpoint must change, not be retried. return jobs.NewProcessingError("datasource.upload_destination_refused", "the object-storage service refuses the source's endpoint", false) case errors.Is(err, connector.ErrCredentialRefused): diff --git a/module/services/accounts/code/pkg/business/export_test.go b/module/services/accounts/code/pkg/business/export_test.go new file mode 100644 index 000000000..f2bb7bcf3 --- /dev/null +++ b/module/services/accounts/code/pkg/business/export_test.go @@ -0,0 +1,7 @@ +package business + +import "accounts/pkg/datasource/connector" + +// ConnectorSource is the envelope's view of a stored source, as the host hands +// it to a registered connector. +func ConnectorSource(source *DatasourceSource) connector.Source { return connectorSource(source) } diff --git a/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md b/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md index d733b87ed..ac2dab24d 100644 --- a/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md +++ b/module/services/accounts/code/pkg/datasource/connector/CONFORMANCE.md @@ -49,9 +49,16 @@ binding and batched read from #108) — over an in-process fake of the service's gRPC API, so every clause is measured on the calls the host makes (`objectstore/connector_conformance_test.go`). The host signs, dials and guards nothing for a bucket; `client.go`, its own store client, is deleted. It is -registered through `connector.Budgeted`, metered per provider credential like -GitHub, and `AdmitNewSource` admits it; the host admits a new source only where -a composed object-storage service can read a bucket (below). +**registered through its service, not as a package**: the host's registry is +built without it, and `pkg/business/datasource_objectstore.go` registers it — +through `Registry.Register`'s own validation, wrapped in `connector.Budgeted` +and metered per provider credential like GitHub — only once the composed +object-storage service resolves and can read a bucket (below). The suite runs +twice: over the service's wire against the package, and against the connector +the registry holds once the provider is registered +(`business/TestObjectStorageAsRegisteredPassesTheConformanceSuite`), on sources +connected through `AddSource`, with the credential the host decrypts per call +and the listings it retains. | Clause | How it is honoured | | --- | --- | @@ -79,11 +86,12 @@ a composed object-storage service can read a bucket (below). - **Every call has a deadline**: 30 seconds per listing page or capability question, five minutes per batched read. A store that answers and then stalls does not hold a sync, and its worker's lease, past it. -- **Admission needs a service that can read a bucket.** At start the host asks - the composed service, unbound, whether it reads a batch; one that does not — - the released 0.0.18 agent, or none composed — leaves object storage - unavailable: the catalog reports `accepts_new_sources: false` and `AddSource` - refuses by name. +- **Registration needs a service that can read a bucket.** At start the host + asks the composed service, unbound, whether it reads a batch; one that does + not — the released 0.0.18 agent — or none composed leaves object storage + unregistered: the catalog does not list it, `AddSource` refuses it by name + (`ErrObjectStorageUnavailable`), and an existing source's sync fails with + the same error. - **The guard and the caps are the service's.** A non-public endpoint is refused there by name (`PERMISSION_DENIED`, reported as the envelope's credential refusal: it is the source's configuration to fix either way). A diff --git a/module/services/accounts/code/work.go b/module/services/accounts/code/work.go index 57714e12b..ffbf6ee3f 100644 --- a/module/services/accounts/code/work.go +++ b/module/services/accounts/code/work.go @@ -359,16 +359,18 @@ func doWork(ctx context.Context) (clean Clean, err error) { // documents module consumes. GITHUB_API_BASE_URL overrides api.github.com for // GitHub Enterprise or tests. service.SetDatasourceConnector(cipher, jobStore, os.Getenv("GITHUB_API_BASE_URL")) - // Object-storage sources are read through the composed object-storage - // service, under a per-call binding of each source's bucket and credential. + // The object-storage datasource provider is resolved from the declared + // object-storage dependency, not compiled in: it is registered only when + // that service is composed and can read a tenant's bucket under a per-call + // binding of the source's bucket and credential. objectStorage, err := dialObjectStorage(ctx) if err != nil { return nil, err } if objectStorage != nil { if err := service.SetObjectStorage(ctx, storagev0.NewObjectStorageClient(objectStorage)); err != nil { - // Not a start failure: the host runs, and object storage is refused - // by name in the catalog, at AddSource and at sync. + // Not a start failure: the host runs without the provider, which + // the catalog does not list and AddSource and sync refuse by name. wool.Get(ctx).Warn("the composed object-storage service cannot read a tenant's bucket; object-storage datasources are unavailable", wool.ErrField(err)) } } @@ -3033,7 +3035,7 @@ func registerMembershipCacheHealth(c *membership.Cache) { // A deployed cell publishes no such token — it runs the service anonymous and // gates callers at the mesh — so an absent token dials without one. A // composition that leaves the service out has no endpoint for it: the host -// then runs without object storage, and an object-storage sync refuses by name. +// then registers no object-storage provider, and a source of it refuses by name. func dialObjectStorage(ctx context.Context) (*grpc.ClientConn, error) { network := codefly.For(ctx).Service("object-storage").Endpoint("grpc").API("grpc").NetworkInstance() if network == nil || network.Address == "" {