server

package
v0.30.38 Latest Latest
Warning

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

 Go to latest Published: Sep 7, 2026 License: Apache-2.0 Imports: 69 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type DxpSweeper added in v0.26.0 

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

DxpSweeper implements gc.Sweeper. Each sweep does two things:

  1. Marks dxp_txn rows still 'active' past their own deadline_ns as 'expired', guarded by the same WHERE status = 'active' CAS discipline markDxpTxnTerminal uses — a row a genuinely in-flight dispatch is about to terminate itself races the sweep on the deadline boundary only in the narrow, benign sense that whichever UPDATE matches first wins; the other affects zero rows.
  2. Purges terminal rows (committed/released/expired) older than retentionSecs, measured from created_at. retentionSecs <= 0 disables purging entirely — tombstones kept forever — matching how this codebase's other retention configs (BlobGCGracePeriodSecs) already treat non-positive as "off," not "purge immediately."

func NewDxpSweeper added in v0.26.0 

func NewDxpSweeper(db *sql.DB, retentionSecs int) *DxpSweeper

NewDxpSweeper creates a DxpSweeper backed by the given database. retentionSecs is how long a terminal instance is kept before this same sweep purges it; see DxpTxnRetentionSecs's own config doc.

func (*DxpSweeper) Sweep added in v0.26.0 

func (d *DxpSweeper) Sweep(ctx context.Context) (gcpkg.Report, error)

Sweep marks stuck-active instances expired, then purges terminal instances past retention — across every tenant in one pair of queries each: dxp_txn is a single global table with tenant_id as a column, not a per-tenant-prefixed one, matching MetaSweeper's own entity_meta sweep shape exactly.

type ErrPromoteInFlight added in v0.26.0 

type ErrPromoteInFlight struct {
	ExistingTicket string
}

func (*ErrPromoteInFlight) Error added in v0.26.0 

func (e *ErrPromoteInFlight) Error() string

type FieldAnalysis added in v0.26.0 

type FieldAnalysis struct {
	Field string `json:"field"`
	// InferredType is a JSON Schema type ("string", "number",
	// "boolean", "object", "array") or "ref" for a field whose values
	// consistently match xolu's own REF wire shape
	// ({"type":"REF","entity":...,"id":...}) -- "ref" is not a JSON
	// Schema type, it's shorthand in this response for
	// {"type":"object","format":"ref"}, which is what actually lands
	// in SuggestedSchema for such a field.
	InferredType string `json:"inferred_type"`
	// Coverage is the fraction of sampled rows where this field was
	// present (0.0-1.0). 1.0 means every sampled row had it -- the
	// basis for suggesting it as required.
	Coverage float64 `json:"coverage"`
	// Confidence is this package's own qualitative read on how much
	// to trust InferredType: "high" for a field with a single
	// consistent type across every observation, "medium" for a
	// pattern-based guess (decimal-like strings), "low" for anything
	// sparse or inconsistent.
	Confidence string `json:"confidence"`
	// SuggestedEnum is set when the field looks closed-set (few
	// distinct values relative to how often it was observed) --
	// present alongside InferredType, not instead of it.
	SuggestedEnum []string `json:"suggested_enum,omitempty"`
	// Note explains anything a caller should know before trusting
	// this field's inference blindly -- e.g. why it was excluded from
	// SuggestedSchema, or why confidence is low.
	Note string `json:"note,omitempty"`
}

FieldAnalysis is one field's inferred shape and the evidence behind it, for a single entity type's schema-suggestion response.

type MetaSweeper 

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

MetaSweeper implements gc.Sweeper. It deletes entity_meta rows whose expires_at is in the past. Fires no events in S3; event dispatch is wired in S9 when the event subscription system lands.

func NewMetaSweeper 

func NewMetaSweeper(db *sql.DB) *MetaSweeper

NewMetaSweeper creates a MetaSweeper backed by the given database.

func (*MetaSweeper) Sweep 

func (m *MetaSweeper) Sweep(ctx context.Context) (gcpkg.Report, error)

Sweep deletes all expired entity_meta rows and returns a gc.Report.

type PromoteJob added in v0.26.0 

type PromoteJob struct {
	Ticket     string
	TenantID   tenant.TenantID
	EntityType string
	Status     PromoteJobStatus
	Result     *PromoteResult
	Failures   []RowValidationFailure
	Error      string
	CreatedAt  time.Time
	FinishedAt time.Time
}

type PromoteJobManager added in v0.26.0 

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

PromoteJobManager tracks strict-promotion jobs in memory, throttled per (tenant, entity type) and bounded in total concurrency -- the same "low priority, never competes unbounded with normal traffic" posture as the export job manager.

func NewPromoteJobManager added in v0.26.0 

func NewPromoteJobManager(maxConcurrent int) *PromoteJobManager

func (*PromoteJobManager) Status added in v0.26.0 

func (m *PromoteJobManager) Status(ticket string) (*PromoteJob, bool)

func (*PromoteJobManager) Submit added in v0.26.0 

func (m *PromoteJobManager) Submit(tenantID tenant.TenantID, entityType string, work func() (*PromoteResult, []RowValidationFailure, error)) (string, error)

Submit starts a strict-promotion job. work is called with no arguments; the caller closes over whatever DB handle/schema it needs, keeping this manager itself free of any dependency on the promotion logic's own signature.

type PromoteJobStatus added in v0.26.0 

type PromoteJobStatus string
const (
	PromoteJobRunning  PromoteJobStatus = "running"
	PromoteJobComplete PromoteJobStatus = "complete"
	PromoteJobFailed   PromoteJobStatus = "failed"
	// PromoteJobRejected is distinct from Failed: Failed means
	// something went wrong (a storage error, a bug); Rejected means
	// strict promotion worked exactly as designed and correctly
	// declined to promote because not every row validated -- the
	// caller's data or schema needs attention, not this code.
	PromoteJobRejected PromoteJobStatus = "rejected"
)

type PromoteResult added in v0.26.0 

type PromoteResult struct {
	MigratedRows int  `json:"migrated_rows"`
	AutoInferred bool `json:"auto_inferred"`
}

type RowValidationFailure added in v0.26.0 

type RowValidationFailure struct {
	ID     int      `json:"id"`
	Errors []string `json:"errors"`
}

RowValidationFailure is one row that failed validation against a candidate schema during strict promotion.

type SchemaSuggestion added in v0.26.0 

type SchemaSuggestion struct {
	EntityType string `json:"entity_type"`
	// SampledRows/TotalRows: inference always runs over a bounded
	// sample (defaultSampleSize), never necessarily every row --
	// both counts are reported so a caller can judge how
	// representative the sample likely was.
	SampledRows int `json:"sampled_rows"`
	TotalRows   int `json:"total_rows"`
	// SuggestedSchema is ready to hand to DefineEntitySchema/Promote
	// as-is, or to edit first. additionalProperties is deliberately
	// left unset (permissive), not forced false -- see this file's own
	// header comment.
	SuggestedSchema map[string]interface{} `json:"suggested_schema"`
	// FieldAnalysis explains the reasoning behind SuggestedSchema,
	// field by field, including fields that were EXCLUDED from it
	// (inconsistent types) so nothing is silently dropped without
	// explanation.
	FieldAnalysis []FieldAnalysis `json:"field_analysis"`
}

SchemaSuggestion is the full response for one entity type.

type Server 

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

Server represents the HTTP server

func New 

func New(
	cfg *config.Config,
	store storage.Store,
	cache cache.Cache,
	g graph.Graph,
	validator validation.Validator,
	logger zerolog.Logger,
) *Server

New creates a new server instance

func (*Server) BlobSamplerFor 

func (s *Server) BlobSamplerFor(tenantID tenant.TenantID) *blob.UsageSampler

BlobSamplerFor returns the UsageSampler for a tenant, or nil when blobs are disabled, the tenant's store is not open, or the sampler interval is zero. Exposed for tests that need to call ForceResample() deterministically.

func (*Server) BlobStoreForTest 

func (s *Server) BlobStoreForTest(tenantID tenant.TenantID) (*blob.Store, error)

BlobStoreForTest opens (if needed) and returns a tenant's blob store via the manager. Exposed for tests that need to seed blobs through the same store instance the server uses, so the per-tenant sampler is created and warm.

func (*Server) CalManagerForTest 

func (s *Server) CalManagerForTest() *cal.Manager

CalManagerForTest exposes the calendar manager for HTTP-level tests that need to seed calendars and bookings through the same manager the handlers use. Returns nil when the cal subsystem is disabled.

Introduced in v0.14.8 alongside the T-18 HTTP test suite. Not part of the public API — the ForTest suffix marks it as a test-only accessor.

func (*Server) HandleTSAggregate 

func (s *Server) HandleTSAggregate(w http.ResponseWriter, r *http.Request)

HandleTSAggregate computes an aggregate over a numeric field. 

POST /api/v1/tenant/{tenant_id}/ts/aggregate

func (*Server) HandleTSAppend 

func (s *Server) HandleTSAppend(w http.ResponseWriter, r *http.Request)

HandleTSAppend appends a single event. 

POST /api/v1/tenant/{tenant_id}/ts/events

func (*Server) HandleTSBatchAppend 

func (s *Server) HandleTSBatchAppend(w http.ResponseWriter, r *http.Request)

HandleTSBatchAppend appends a batch of events atomically. 

POST /api/v1/tenant/{tenant_id}/ts/events/batch

func (*Server) HandleTSDefineRollup 

func (s *Server) HandleTSDefineRollup(w http.ResponseWriter, r *http.Request)

HandleTSDefineRollup creates a rollup definition on a source timeline. 

POST /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/def

func (*Server) HandleTSDefineTimeline 

func (s *Server) HandleTSDefineTimeline(w http.ResponseWriter, r *http.Request)

HandleTSDefineTimeline defines or updates a timeline. 

POST /api/v1/tenant/{tenant_id}/ts/timelines

func (*Server) HandleTSDeleteRollup 

func (s *Server) HandleTSDeleteRollup(w http.ResponseWriter, r *http.Request)

HandleTSDeleteRollup removes a rollup definition and stops its worker. 

DELETE /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/{rollup_id}

func (*Server) HandleTSDeleteTimeline 

func (s *Server) HandleTSDeleteTimeline(w http.ResponseWriter, r *http.Request)

HandleTSDeleteTimeline removes a timeline definition together with its event data and rollups (the inverse of define; distinct from DeleteTimelineData, which keeps the definition). When rollup cascade is disabled and the timeline still has rollups, the store returns ErrTSRollupDestInUse and this responds 409 so the caller knows to remove the rollups first; an unknown timeline is 404. 

DELETE /api/v1/tenant/{tenant_id}/ts/tl/{timeline_id}

func (*Server) HandleTSDeleteTimelineData 

func (s *Server) HandleTSDeleteTimelineData(w http.ResponseWriter, r *http.Request)

HandleTSDeleteTimelineData removes all events from a timeline. 

DELETE /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/data

func (*Server) HandleTSFullAggregate 

func (s *Server) HandleTSFullAggregate(w http.ResponseWriter, r *http.Request)

func (*Server) HandleTSGetRetention 

func (s *Server) HandleTSGetRetention(w http.ResponseWriter, r *http.Request)

HandleTSGetRetention returns the retention configuration. 

GET /api/v1/tenant/{tenant_id}/ts/retention

func (*Server) HandleTSGetRollup 

func (s *Server) HandleTSGetRollup(w http.ResponseWriter, r *http.Request)

HandleTSGetRollup returns a specific rollup definition. 

GET /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/{rollup_id}

func (*Server) HandleTSGetTimeline 

func (s *Server) HandleTSGetTimeline(w http.ResponseWriter, r *http.Request)

HandleTSGetTimeline returns a single timeline. 

GET /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}

func (*Server) HandleTSLatest 

func (s *Server) HandleTSLatest(w http.ResponseWriter, r *http.Request)

HandleTSLatest returns the N most recent events. 

GET /api/v1/tenant/{tenant_id}/ts/events/latest

func (*Server) HandleTSListRollups 

func (s *Server) HandleTSListRollups(w http.ResponseWriter, r *http.Request)

HandleTSListRollups lists all rollup definitions for a source timeline. 

GET /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/list

func (*Server) HandleTSListTimelines 

func (s *Server) HandleTSListTimelines(w http.ResponseWriter, r *http.Request)

HandleTSListTimelines returns all defined timelines. 

GET /api/v1/tenant/{tenant_id}/ts/timelines

func (*Server) HandleTSPatchRetention 

func (s *Server) HandleTSPatchRetention(w http.ResponseWriter, r *http.Request)

func (*Server) HandleTSProvision 

func (s *Server) HandleTSProvision(w http.ResponseWriter, r *http.Request)

HandleTSProvision provisions timeseries for a tenant. 

POST /api/v1/tenant/{tenant_id}/ts/provision

func (*Server) HandleTSPurgeTimelineRange 

func (s *Server) HandleTSPurgeTimelineRange(w http.ResponseWriter, r *http.Request)

HandleTSPurgeTimelineRange removes events in a time range from a timeline. 

POST /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/data/purge

func (*Server) HandleTSQueryRange 

func (s *Server) HandleTSQueryRange(w http.ResponseWriter, r *http.Request)

HandleTSQueryRange returns events in a time range. 

GET /api/v1/tenant/{tenant_id}/ts/events

func (*Server) HandleTSQueryRangePost 

func (s *Server) HandleTSQueryRangePost(w http.ResponseWriter, r *http.Request)

HandleTSQueryRangePost is the POST equivalent of HandleTSQueryRange. It accepts the same parameters as a JSON body instead of query-string values, which is more ergonomic for complex queries and avoids URL-length limits. 

POST /api/v1/tenant/{tenant_id}/ts/query/range

Request body: 

{
  "timeline": 1,
  "dims":     [42],
  "from":     "2026-01-01T00:00:00Z",
  "to":       "2026-01-02T00:00:00Z",
  "limit":    1000,   // optional; default 1000, max TSMaxQueryEvents
  "order":    "asc"   // optional; "asc" (default) or "desc"
}

func (*Server) HandleTSRangeAggregate 

func (s *Server) HandleTSRangeAggregate(w http.ResponseWriter, r *http.Request)

func (*Server) HandleTSRollupParent 

func (s *Server) HandleTSRollupParent(w http.ResponseWriter, r *http.Request)

HandleTSRollupParent returns the rollup definition for which this timeline is the destination — i.e. its parent in the rollup tree. 

GET /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/parent

func (*Server) HandleTSRollupStatus 

func (s *Server) HandleTSRollupStatus(w http.ResponseWriter, r *http.Request)

HandleTSRollupStatus returns the operational status of a rollup worker. 

GET /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/{rollup_id}/status

func (*Server) HandleTSRollupTree 

func (s *Server) HandleTSRollupTree(w http.ResponseWriter, r *http.Request)

HandleTSRollupTree returns the full rollup tree for the tenant. 

GET /api/v1/tenant/{tenant_id}/ts/rollup/tree

func (*Server) HandleTSRunRollup 

func (s *Server) HandleTSRunRollup(w http.ResponseWriter, r *http.Request)

HandleTSRunRollup manually triggers a rollup execution for the given range. If cascade is true in the request body, all descendant rollup definitions are also run for the corresponding time windows, in source→destination order. Workers are started for this definition and all cascaded descendants. 

POST /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/rollup/{rollup_id}/run

func (*Server) HandleTSStats 

func (s *Server) HandleTSStats(w http.ResponseWriter, r *http.Request)

HandleTSStats returns store-level diagnostics. 

GET /api/v1/tenant/{tenant_id}/ts/stats

func (*Server) HandleTSSyncGet 

func (s *Server) HandleTSSyncGet(w http.ResponseWriter, r *http.Request)

HandleTSSyncGet returns the current nosync setting for a timeline. 

GET /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/sync

func (*Server) HandleTSSyncOff 

func (s *Server) HandleTSSyncOff(w http.ResponseWriter, r *http.Request)

HandleTSSyncOff enables nosync mode for a timeline (NoSync=true). AppendBatch will no longer wait for WAL fsync before returning. Data loss is possible if the process crashes before the OS flushes the WAL to disk. The loss window is bounded by the kernel dirty-page writeback interval, typically under one second. 

POST /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/sync/off

func (*Server) HandleTSSyncOn 

func (s *Server) HandleTSSyncOn(w http.ResponseWriter, r *http.Request)

HandleTSSyncOn restores synchronous write mode for a timeline (NoSync=false). AppendBatch will again wait for WAL fsync before returning. This is the default mode and provides crash durability. 

POST /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}/sync/on

func (*Server) HandleTSTimelineStats 

func (s *Server) HandleTSTimelineStats(w http.ResponseWriter, r *http.Request)

HandleTSTimelineStats returns per-timeline diagnostics. 

GET /api/v1/tenant/{tenant_id}/ts/stats/{timeline_id}

func (*Server) HandleTSUpdateTimeline 

func (s *Server) HandleTSUpdateTimeline(w http.ResponseWriter, r *http.Request)

HandleTSUpdateTimeline updates a timeline's mutable fields. 

PATCH /api/v1/tenant/{tenant_id}/ts/timelines/{timeline_id}

func (*Server) Handler 

func (s *Server) Handler() http.Handler

Handler returns the HTTP handler (useful for testing)

func (*Server) IsAdaptedForTest added in v0.30.23 

func (s *Server) IsAdaptedForTest(tenantID tenant.TenantID, entity string) (bool, error)

IsAdaptedForTest reports whether entity has an adapted table on the ACTUAL store instance a given tenant's own requests use -- not s.storage, which is what naive test code would check first and get a wrong answer from for any non-zero tenant. Exists for XOT178's own verification: confirms registerAdaptedEverywhere/replayAdaptedSchemas genuinely reach the right store, not just that they ran without error.

func (*Server) LocStoreForTest added in v0.26.0 

func (s *Server) LocStoreForTest(ctx context.Context, tenantID tenant.TenantID) (*loc.Store, error)

LocStoreForTest exposes locStore's own tenantID-keyed core for HTTP-level and dxp-integration tests — the same ForTest pattern as BlobStoreForTest/CalManagerForTest.

func (*Server) MarkReady 

func (s *Server) MarkReady()

MarkReady sets the server as ready. This is called automatically by Start(), but can be called manually in test setups that use httptest.NewServer instead of Start().

func (*Server) ObjStoreForTest added in v0.26.0 

func (s *Server) ObjStoreForTest(ctx context.Context, tenantID tenant.TenantID) (*obj.Store, error)

ObjStoreForTest mirrors LocStoreForTest's own purpose exactly.

func (*Server) S3Handler 

func (s *Server) S3Handler() http.Handler

S3Handler returns an http.Handler for the S3-compatible API surface. Used in tests to wire the S3 router to an httptest.Server without needing a real TCP port. Returns nil when the blob store is not initialised.

func (*Server) SetGraphQueryCacheTTL 

func (s *Server) SetGraphQueryCacheTTL(seconds int)

── Graph query result cache ──────────────────────────────────────────────────

SetGraphQueryCacheTTL sets the whole-query result cache TTL at runtime. Primarily used by tests that need a short TTL without rebuilding the server. A value of 0 disables query result caching.

func (*Server) SetTSManager 

func (s *Server) SetTSManager(m timeseries.Manager)

SetTSManager replaces the server's timeseries Manager. Intended for testing only: it allows injecting a fake or failing manager after construction so that failure paths (e.g. XOLU-CM016) can be exercised deterministically. Must not be called concurrently with request handling.

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the HTTP server, allowing in-flight requests to complete within the given context deadline.

func (*Server) Start 

func (s *Server) Start() error

Start starts the HTTP server

func (*Server) Stop 

func (s *Server) Stop()

Stop stops the server and cleans up resources

func (*Server) TSManager 

func (s *Server) TSManager() timeseries.Manager

TSManager returns the server's timeseries Manager. Returns nil when timeseries is disabled. Exposed primarily for testing — production code should not replace the manager after the server has started serving requests.

func (*Server) TenantIDForTest added in v0.26.0 

func (s *Server) TenantIDForTest(name string) (tenant.TenantID, bool)

TenantIDForTest resolves a tenant name (e.g. "default") to its numeric tenant.TenantID — the same lookup resolveNumericTenant's own registry uses internally, exposed for tests that need to construct a node id (tenantID.NodeID) matching exactly what server-side code would produce, rather than guessing at the tenant-prefix format (T-123's own graph-mirroring proof needed this: the legacy /graph/neighbors endpoint is deliberately unscoped, "sees all tenant nodes" per its own route comment, so a wrong guess here silently queries for a node that was never mirrored under that exact key, not a genuine mirror failure).

func (*Server) TenantRegistry 

func (s *Server) TenantRegistry() *tenant.Registry

TenantRegistry returns the server's tenant registry. This is primarily useful for pre-registering tenants in strict mode before starting the server.

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.