From 1b9aa45a21b43cdb0f2626d33b0b9c9d51c65be6 Mon Sep 17 00:00:00 2001 From: DeVaris Brown Date: Sun, 5 Jul 2026 18:00:52 -0700 Subject: [PATCH] docs: godoc and package-doc pass Add root and schema package doc.go files and document the highest-value author-facing contracts: the Processor lifecycle, ProcessedRecord error propagation semantics, and the standalone (WASM) vs built-in hosting model. Co-Authored-By: Claude Opus 4.8 (1M context) --- doc.go | 133 ++++++++++++++++++++++++++++++++++++++++++++ errors.go | 4 ++ processor_func.go | 6 ++ schema/doc.go | 35 ++++++++++++ schema/in_memory.go | 6 ++ schema/schema.go | 21 ++++++- 6 files changed, 204 insertions(+), 1 deletion(-) create mode 100644 doc.go create mode 100644 schema/doc.go diff --git a/doc.go b/doc.go new file mode 100644 index 0000000..918f9e2 --- /dev/null +++ b/doc.go @@ -0,0 +1,133 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package sdk is the Go SDK for building Conduit processors. A processor +// receives records flowing through a pipeline, transforms them, and returns the +// result. Authors implement the [Processor] interface (or adapt a function with +// [NewProcessorFunc]) and the SDK handles the rest: configuration parsing, +// schema encode/decode middleware, and the plumbing that connects the processor +// to the Conduit engine. +// +// # Standalone vs. built-in processors +// +// The same [Processor] implementation can run in two ways: +// +// - Standalone: the processor is compiled to a WebAssembly module +// (GOOS=wasip1 GOARCH=wasm) and executed by Conduit in a wazero runtime. +// The module's main function calls [Run], which becomes the entry point. +// This is the default and recommended mode — it isolates the processor from +// the engine and lets it be distributed as a single portable binary. +// - Built-in: the processor is compiled natively into a custom Conduit +// build. This avoids the WebAssembly boundary (and its per-call +// serialization cost) at the price of coupling the processor to the engine +// binary. Built-in processors do not call [Run]; the engine invokes the +// [Processor] methods directly. +// +// Author code is identical in both modes. Only the entry point and build +// constraints differ, so write to the [Processor] contract and let the build +// target decide how the processor is hosted. +// +// # Implementing a Processor +// +// Embed [UnimplementedProcessor] in your type. It provides no-op implementations +// of the optional methods and satisfies the unexported marker method that keeps +// the interface closed, so adding a method to [Processor] in a later release is +// not a breaking change for existing processors: +// +// type myProcessor struct { +// sdk.UnimplementedProcessor +// cfg myConfig +// } +// +// func (p *myProcessor) Specification() (sdk.Specification, error) { ... } +// func (p *myProcessor) Configure(ctx context.Context, cfg config.Config) error { ... } +// func (p *myProcessor) Process(ctx context.Context, recs []opencdc.Record) []sdk.ProcessedRecord { ... } +// +// # Lifecycle +// +// The runtime calls a processor's methods in a fixed order, and (for a single +// processor instance) never concurrently — the standalone command loop in [Run] +// processes one command at a time. A processor therefore does not need to guard +// its own fields against concurrent access by the SDK, but it must not assume +// any parallelism either. +// +// 1. Specification — called to discover the processor's name, version, and +// configuration parameters. Must be side-effect free; it may be called +// before Configure and without any configuration. +// 2. Configure — called once with the user's configuration. Validate and store +// it here. Do not open connections or start background work; that is Open's +// job. See [ParseConfig] for turning the raw config map into a typed struct. +// 3. Open — called once after Configure. Acquire resources and start any +// background work here. +// 4. Process — called repeatedly, once per incoming batch, until shutdown. See +// the record-handling contract below. +// 5. Teardown — called once when the pipeline is shutting down. No other method +// is called after Teardown returns; the processor is then discarded. Release +// everything Open acquired. +// +// Process may be called more than once with the same records (for example after +// a restart when records were not flushed downstream), so processing must be +// idempotent. +// +// # Record handling and error propagation +// +// Process receives a batch of [opencdc.Record] values and returns a +// [ProcessedRecord] for each. The returned slice is positional: the result at +// index i is the outcome of the input record at index i. Each input record may +// carry raw or structured data in its key and payload; a processor that reads +// structured fields should enable the schema-decode middleware (see below) or +// handle both shapes. +// +// The concrete [ProcessedRecord] type an author returns decides how the record +// continues through the pipeline: +// +// - [SingleRecord] — the transformed record continues downstream. This is the +// common case. +// - [MultiRecord] — the record is split into zero or more records. Returning +// an empty MultiRecord is equivalent to [FilterRecord]; returning one record +// is equivalent to [SingleRecord]. +// - [FilterRecord] — the record is acknowledged and dropped from the pipeline. +// Use this to intentionally discard records; it is not an error. +// - [ErrorRecord] — processing failed. The record is nacked and handled +// according to the pipeline's error policy (for example routed to a dead- +// letter queue or halting the pipeline). Returning an ErrorRecord is the +// only way to signal a per-record failure — a processor must not drop a +// record it could not process, or at-least-once delivery is violated. +// +// Because filtering, splitting, and failing are all expressed through the +// return value rather than through the process's exit or a returned error, +// Process itself does not return an error: a batch always produces a result for +// every record it accounts for. +// +// # Middleware +// +// [DefaultProcessorMiddleware] wraps every processor run through [Run] with +// schema decode and encode middleware. Decode middleware fetches the schema +// referenced in a record's metadata and turns raw key/payload bytes into +// [opencdc.StructuredData] before Process sees them; encode middleware reverses +// that afterwards, so a processor can operate on structured data without dealing +// with schema resolution. A processor tunes this via [Processor.MiddlewareOptions] +// (see [ProcessorWithSchemaDecodeConfig] and [ProcessorWithSchemaEncodeConfig]). +// +// # Neighboring packages +// +// - github.com/conduitio/conduit-commons/opencdc defines the Record type that +// flows through Process. +// - github.com/conduitio/conduit-commons/config defines the configuration and +// parameter types used in Specification and Configure. +// - The schema subpackage is the author-facing API for creating and fetching +// schemas from within a processor. +// - The wasm and pprocutils subpackages are engine plumbing and are not meant +// to be imported by processor authors. +package sdk diff --git a/errors.go b/errors.go index 1dcd1ca..6499a73 100644 --- a/errors.go +++ b/errors.go @@ -23,5 +23,9 @@ var ( "this action, please check the source code of the processor and make sure " + "all required processor methods are implemented") + // ErrFilterRecord is a sentinel error a function passed to + // [NewProcessorFunc] can return to filter a record out of the pipeline + // instead of failing it. The record is acked and dropped ([FilterRecord]), + // not nacked. Returning any other error yields an [ErrorRecord]. ErrFilterRecord = errors.New("filter out this record") ) diff --git a/processor_func.go b/processor_func.go index 4e94c57..94509cd 100644 --- a/processor_func.go +++ b/processor_func.go @@ -47,6 +47,12 @@ func NewProcessorFunc(specs Specification, f func(context.Context, opencdc.Recor func (f ProcessorFunc) Specification() (Specification, error) { return f.specs, nil } +// Process applies the wrapped function to each record in order. A record for +// which the function returns [ErrFilterRecord] becomes a [FilterRecord]; any +// other error becomes an [ErrorRecord] and processing stops there, so the +// returned slice is truncated at the first failing record and is shorter than +// the input. Records before the error are returned as [SingleRecord] values; +// records after it are left for Conduit to reprocess. func (f ProcessorFunc) Process(ctx context.Context, records []opencdc.Record) []ProcessedRecord { outRecs := make([]ProcessedRecord, len(records)) for i, inRec := range records { diff --git a/schema/doc.go b/schema/doc.go new file mode 100644 index 0000000..fc1e5a8 --- /dev/null +++ b/schema/doc.go @@ -0,0 +1,35 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package schema is the processor-facing API for creating and fetching schemas. +// Processors use it to register the schema of records they emit and to look up +// the schema of records they receive, keeping schema identity consistent across +// a pipeline. +// +// The package exposes two entry points, [Get] and [Create], both backed by the +// package-level [SchemaService]. The service a processor talks to depends on how +// it is hosted: +// +// - Standalone (WebAssembly): the engine replaces [SchemaService] at startup +// with an implementation that forwards calls to Conduit's schema registry +// over the host boundary, so schemas are shared with the rest of the +// pipeline. +// - Built-in / tests: the default [SchemaService] is an in-process, +// [NewInMemoryService]-backed store wrapped in a cache. It has no +// persistence and is not shared with a real registry — useful for unit +// tests, not for cross-processor schema sharing. +// +// Get and Create results are cached, so repeated lookups of the same schema do +// not cross the host boundary again. +package schema diff --git a/schema/in_memory.go b/schema/in_memory.go index bbf08d4..c199873 100644 --- a/schema/in_memory.go +++ b/schema/in_memory.go @@ -23,6 +23,11 @@ import ( "github.com/conduitio/conduit-processor-sdk/pprocutils" ) +// InMemoryService is a non-persistent [pprocutils.SchemaService] that keeps all +// schemas in memory. It is the default backing store for tests and built-in +// processors; it is safe for concurrent use but its contents are lost when the +// process exits and are not shared with a real schema registry. Versions per +// subject start at 1 and increment on each [InMemoryService.CreateSchema] call. type InMemoryService struct { // schemas is a map of schema subjects to all the versions of that schema // versioning starts at 1, newer versions are appended to the end of the versions slice. @@ -33,6 +38,7 @@ type InMemoryService struct { idSequence int } +// NewInMemoryService returns an empty [InMemoryService] ready for use. func NewInMemoryService() *InMemoryService { return &InMemoryService{ schemas: make(map[string][]schema.Schema), diff --git a/schema/schema.go b/schema/schema.go index 3872d83..dd9c430 100644 --- a/schema/schema.go +++ b/schema/schema.go @@ -22,16 +22,31 @@ import ( "github.com/conduitio/conduit-processor-sdk/pprocutils" ) +// TypeAvro is the Avro schema type. It is currently the only type accepted by +// [Create]. const TypeAvro = schema.TypeAvro +// SchemaService is the service backing [Get] and [Create]. In a standalone +// (WebAssembly) processor the engine overwrites this at startup with a client +// for Conduit's schema registry; otherwise it defaults to an in-process, +// cache-wrapped in-memory store. Replace it in tests to stub schema resolution. var SchemaService pprocutils.SchemaService = newCachedSchemaService(NewInMemoryService()) var ( + // ErrSubjectNotFound is returned by [Get] when no schema exists for the + // requested subject. ErrSubjectNotFound = pprocutils.ErrSubjectNotFound + // ErrVersionNotFound is returned by [Get] when the subject exists but the + // requested version does not. ErrVersionNotFound = pprocutils.ErrVersionNotFound - ErrInvalidSchema = pprocutils.ErrInvalidSchema + // ErrInvalidSchema is returned by [Create] when the supplied bytes are not a + // valid schema of the requested type. + ErrInvalidSchema = pprocutils.ErrInvalidSchema ) +// Get fetches the schema registered under the given subject and version. +// Versions start at 1. It returns [ErrSubjectNotFound] or [ErrVersionNotFound] +// (wrapped) if the schema is not registered. func Get(ctx context.Context, subject string, version int) (schema.Schema, error) { resp, err := SchemaService.GetSchema(ctx, pprocutils.GetSchemaRequest{ Subject: subject, @@ -43,6 +58,10 @@ func Get(ctx context.Context, subject string, version int) (schema.Schema, error return resp.Schema, nil } +// Create registers a new version of the schema for the given subject and +// returns it with its assigned ID and version. Each call to Create appends a new +// version; there is no deduplication of identical bytes. Only [TypeAvro] is +// currently accepted — other types return [ErrInvalidSchema]. func Create(ctx context.Context, typ schema.Type, subject string, bytes []byte) (schema.Schema, error) { resp, err := SchemaService.CreateSchema(ctx, pprocutils.CreateSchemaRequest{ Subject: subject,