connection

package
v2.5.0-alpha.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 13, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultInitialInterval = 100 * time.Millisecond
	DefaultMaxInterval     = 10 * time.Second
)
View Source 
const ConnectionRecovering = "recovering"

ConnectionRecovering is the runtime-reconnect state: a previously connected logical connection lost its transport and (for self-recovering clients) is re-establishing it, or (for pool-recovered clients) is waiting for the Pool recovery worker. It shares the disconnected generation: waiters stay parked, they are not woken and re-parked. Defined here until it is promoted to the contract api on the next contract release; the wire value is fixed as "recovering".

View Source 
const (
	LblName = "name"
)

Variables

View Source 
var (
	// ErrConnectionClosed reports that a waiter was woken by connection
	// lifecycle termination (Drop/Update/zero-ref/shutdown) rather than
	// by its own caller context. It must never be confused with
	// caller cancellation (ctx.Err()).
	ErrConnectionClosed = errors.New("connection lifecycle closed")
	// ErrConnectionRemoving reports that the requested key is owned by
	// teardown. Callers apply operation-specific policy: RequireExisting
	// fetches and named creates fail fast with it, while anonymous
	// fetches wait for cleanup and retry the lookup.
	ErrConnectionRemoving = errors.New("connection is being removed")
)
View Source 
var ConnStatusGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
	Namespace: "kuiper",
	Subsystem: "conn_status",
	Name:      "gauge",
	Help:      "gauge of connection status",
}, []string{LblName})

Functions

func ConnectionHealthProbeJob 

func ConnectionHealthProbeJob(ctx context.Context)

ConnectionHealthProbeJob periodically verifies the health of named connections that opt into modules.PeriodicHealthChecker. It is failure discovery only: a failed HealthCheck flips connected to disconnected (opening a new readiness generation for WaitReady waiters) and hands the episode to the per-Meta recovery worker via nudgeRecovery. It never recovers, never reconnects — recovery belongs to the worker. It runs independent of the status Patrol: the Patrol stays a pure-read metrics job that a slow probe can never stall.

func ConsumerRefID 

func ConsumerRefID(ctx api.StreamContext) string

ConsumerRefID derives the stable Source/Sink consumer identity (ruleID + opID + instanceID) for the current context. Connect must save nothing: the returned Lease already binds it; Release detaches it.

func CreateMockConnection 

func CreateMockConnection(ctx api.StreamContext) modules.Connection

func DropNameConnection 

func DropNameConnection(ctx api.StreamContext, selId string) error

func InitConnectionManager 

func InitConnectionManager(ctx context.Context)

func InitConnectionManager4Test 

func InitConnectionManager4Test() error

func LookupRefID 

func LookupRefID(ctx api.StreamContext) (string, bool)

LookupRefID reads the framework-injected lookup resource identity. The second return value reports whether a non-empty identity was present.

func NewExponentialBackOffWithMaxElapsedTime added in v2.2.8 

func NewExponentialBackOffWithMaxElapsedTime(maxElapsedTime time.Duration) *backoff.ExponentialBackOff

func PatrolConnectionStatusJob added in v2.2.0 

func PatrolConnectionStatusJob(ctx context.Context)

func ReloadNamedConnection 

func ReloadNamedConnection() error

ReloadNamedConnection is called when server starts. It initializes all stored named connections

func ResolveConnectionKey 

func ResolveConnectionKey(props map[string]any, anonymous string) (key string, requireExisting bool)

ResolveConnectionKey derives the explicit pool identity for one fetch from connector-local material: any non-empty string props["connectionSelector"] selects the named connection (RequireExisting=true, the fetch only attaches); otherwise the caller-provided anonymous key is used and the fetch may create. An empty selector counts as absent. The Pool treats the returned key as opaque.

func WithLookupRefID 

func WithLookupRefID(ctx *topoContext.DefaultContext, refID string) *topoContext.DefaultContext

WithLookupRefID injects the framework-owned lookup resource identity into ctx. Called by lookup.CreateInstance before LookupSource.Connect; the source only reads and stores the value, never derives it.

The parameter is deliberately *topoContext.DefaultContext, the only context the lookup framework ever creates: there is no production need to inject into an arbitrary api.StreamContext, and wrapping one would silently drop its rule/op/instance identity.

Types

type ConnectionLease 

type ConnectionLease struct {
	// contains filtered or unexported fields
}

ConnectionLease is the per-consumer handle on a shared logical connection. Each successful fetch mints exactly one Lease binding one attachment (connectionKey + refID + attachment token) to the Meta-shared internal handle, so a consumer can never release the wrong reference: Release only detaches when the token still matches the recorded attachment. A Lease from an earlier same-ref attachment is stale: its Release is a no-op. Business methods delegate to the shared handle; only Release is per-Lease state. A nil Lease is safe to use: business methods behave as if the connection were never acquired, and Release is a no-op.

Attachment vs handle semantics (deliberate): only Release is per-attachment. Wait, WaitReady, Status and ReportSuspectedFailure address the shared logical connection, so they stay usable after Release and are intentionally not token-gated. A suspect is transport-health signal, not an ownership mutation — even a superseded holder that just failed real I/O reports a true signal, and the Pool worker verifies (Ping) before acting on it, so a spurious report costs at most one verification.

func CreateNamedConnection 

func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[string]any) (*ConnectionLease, error)

CreateNamedConnection creates a named connection without attaching a consumer reference: the returned Lease holds no ref, so Release on it is a no-op. It exists so API handlers and tests can observe the shared handle (Wait/Status) without owning a reference.

func FetchConnectionWithOptions 

func FetchConnectionWithOptions(ctx api.StreamContext, opts FetchOptions) (*ConnectionLease, error)

FetchConnectionWithOptions is the explicit-identity fetch path. Callers must canonicalize props/key and compute RefID before calling; Pool treats ConnectionKey as an opaque identity. Every successful fetch mints one ConnectionLease binding this caller's reference; Release it to detach.

func UpdateConnection 

func UpdateConnection(ctx api.StreamContext, id, typ string, props map[string]any) (*ConnectionLease, error)

func (*ConnectionLease) ConnectionKey 

func (l *ConnectionLease) ConnectionKey() string

ConnectionKey returns the pool key this Lease is attached to.

func (*ConnectionLease) IsInitialized 

func (l *ConnectionLease) IsInitialized() bool

IsInitialized reports whether the initial connection worker has published its first result: false while the first Dial is still in flight (a lookup against it reports not-ready once instead of parking), true once a handle or an error is available.

func (*ConnectionLease) Release 

func (l *ConnectionLease) Release(ctx api.StreamContext) error

Release detaches this Lease's own attachment exactly once; repeated calls are a no-op returning nil. A Lease holding no reference (named creation, or nil) releases nothing. A stale Lease — attachment superseded by a reattach — releases nothing: the current holder is untouched.

func (*ConnectionLease) ReportSuspectedFailure 

func (l *ConnectionLease) ReportSuspectedFailure()

ReportSuspectedFailure reports one failed business I/O against the pooled connection. See the shared handle for the full contract.

func (*ConnectionLease) Status 

func (l *ConnectionLease) Status() (string, string)

Status reports the last-known connection state. Pure read.

func (*ConnectionLease) Wait 

func (l *ConnectionLease) Wait(ctx api.StreamContext) (modules.Connection, error)

Wait blocks until the logical connection is first usable. See the shared handle for the full contract.

func (*ConnectionLease) WaitReady 

func (l *ConnectionLease) WaitReady(ctx api.StreamContext) error

WaitReady blocks until the pooled connection is internally ready. See the shared handle for the full contract.

type FetchOptions 

type FetchOptions struct {
	// ConnectionKey is the opaque logical-connection identity. For named
	// connections it is the connectionSelector; for anonymous ones it is
	// the connector-provided canonical key (e.g. resolved SQL DB URL).
	ConnectionKey string
	// RefID identifies the holding consumer. Source/Sink use the
	// rule+op+instance identity; lookup tables use the framework-injected
	// lookup resource identity. It must be saved by the consumer at
	// Connect time and passed back verbatim at detach time.
	RefID string
	// RequireExisting marks a fetch that may only attach to an already
	// existing logical connection: a missing ConnectionKey is an error
	// and anonymous creation is never performed. Used for
	// connectionSelector/named references. False allows creating an
	// anonymous Meta when the key is absent. This describes the fetch
	// mode, not the Meta lifecycle type (Meta.Named).
	RequireExisting bool
	Type            string
	Props           map[string]any
	// StatusHandler receives connection status changes for this ref.
	// Deliveries run serially on the per-Meta dispatcher, in enqueue
	// order, each exactly once. Contract: return promptly and stay
	// memory-only (no I/O, no blocking or waiting) — a slow handler
	// delays later deliveries on the same Meta and, at teardown, the
	// stop-path drain. Unlike the pre-dispatcher design, handlers may
	// synchronously re-enter Pool operations: producers never hold
	// the ordering lock across an invocation, so re-entry cannot
	// self-deadlock. See Meta.eventMu.
	StatusHandler api.StatusChangeHandler
}

FetchOptions carries the explicit connection identity for the new fetch path (DESIGN §8, A1a). ConnectionKey selects which logical connection to share; RefID identifies which consumer holds it. The two must never be mixed: Pool never derives one from the other.

type Manager 

type Manager struct {
	syncx.RWMutex
	// contains filtered or unexported fields
}

type Meta 

type Meta struct {
	ID    string         `json:"id"`
	Typ   string         `json:"typ"`
	Props map[string]any `json:"props"`
	// named means connection is created manually
	Named bool `json:"named"`
	// contains filtered or unexported fields
}

func GetAllConnectionsMeta 

func GetAllConnectionsMeta(forceAll bool) []*Meta

func GetConnectionDetail 

func GetConnectionDetail(_ api.StreamContext, id string) (*Meta, error)

func (*Meta) AddRef 

func (meta *Meta) AddRef(refId string, sc api.StatusChangeHandler) uint64

AddRef registers one consumer reference and mints its attachment token. It is structural only: no status read, no callback, no I/O — safe under the Manager lock. Every call mints a fresh token, even a same-ref reattach that replaces the handler without growing the count: the token identifies this attachment, not just the refID, so a stale Lease from an earlier attachment can never release or observe a later one. The initial state delivery is a separate step (deliverInitial) that runs after the Manager lock is released, so a slow consumer can never stall the Pool. Registration precedes delivery; combined with per-Meta FIFO event order (eventMu) every handler observes each transition exactly once and in order, never concurrently and never inverted. A racing initial delivery may duplicate the latest state (connected, connected) but never reports new-then-old and never swallows a real transition.

func (*Meta) DeRef 

func (meta *Meta) DeRef(refId string) bool

func (*Meta) GetRefCount 

func (meta *Meta) GetRefCount() int

func (*Meta) GetRefNames 

func (meta *Meta) GetRefNames() (result []string)

func (*Meta) GetStatus 

func (meta *Meta) GetStatus() (s string, e string)

GetStatus reports the last-known connection state maintained by the Pool. It is a pure read: it never dials, pings or waits. Freshness comes from asynchronous producers (provider status callbacks, the initial dial loop, the health probe), never from the read itself. Callers must not interpret connected as "probed just now".

func (*Meta) NotifyStatus 

func (meta *Meta) NotifyStatus(status string, s string)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.