Skip to main content

cmd/taisce/ingest.go

cmd/taisce · 378 lines · 17 declarations · source

This file carries the package documentation, rendered on the package page.

Declarations​

const envAPI, envToken​

const (
	envAPI   = "TAISCE_API"
	envToken = "TAISCE_TOKEN"
)

source

type manifest​

type manifest struct {
	Version       int              `json:"version"`
	Document      string           `json:"document"`
	Digest        string           `json:"digest"`
	Bytes         int              `json:"bytes"`
	Role          string           `json:"role"`
	DataSubjectID string           `json:"data_subject_id,omitempty"`
	OccurredAt    *time.Time       `json:"occurred_at,omitempty"`
	SegmentBytes  int              `json:"segment_bytes"`
	Segments      []segmentReceipt `json:"segments"`
	// Refused says why the document stopped, if it did; a manifest with it set lists the segments
	// that were received before the refusal, and a rerun sends them all again as replays.
	Refused string `json:"refused,omitempty"`
}

manifest is what the command writes for one document. Version names the shape so a reader of a manifest written by an older binary knows what it is reading.

source

type segmentReceipt​

type segmentReceipt struct {
	Ordinal        int    `json:"ordinal"`
	ByteStart      int    `json:"byte_start"`
	ByteEnd        int    `json:"byte_end"`
	IdempotencyKey string `json:"idempotency_key"`
	ObservationID  string `json:"observation_id"`
	LogOffset      int64  `json:"log_offset"`
}

source

type ingestOptions​

type ingestOptions struct {
	api           string
	role          string
	dataSubjectID string
	occurredAt    string
	segmentBytes  int
	maxFileBytes  int
	manifestDir   string
	capacityWait  time.Duration
	wait          time.Duration
}

ingestOptions are the flags, parsed once and passed rather than read from globals.

source

func ingestCommand​

func ingestCommand(ctx context.Context, args []string, out io.Writer) error

source

func ingestOne​

func ingestOne(ctx context.Context, client *apiClient, path string, o ingestOptions, defaultTime *time.Time) (manifest, error)

ingestOne reads, cuts and sends one document, and returns the manifest so far with the error that stopped it, if one did. A refusal by the package leaves an empty segment list; a refusal by the API leaves the receipts received before it.

source

func writeManifest​

func writeManifest(dir string, m manifest) error

source

type observeBody​

type observeBody struct {
	IdempotencyKey string           `json:"idempotency_key"`
	DataSubjectID  string           `json:"data_subject_id,omitempty"`
	OccurredAt     *time.Time       `json:"occurred_at,omitempty"`
	Messages       []observeMessage `json:"messages"`
}

The wire shapes this command uses, written here rather than imported from the API package: a client that shares the server's types cannot notice the server changing them.

source

type observeMessage​

type observeMessage struct {
	GroupOrdinal int    `json:"group_ordinal"`
	Role         string `json:"role"`
	Content      string `json:"content"`
}

source

type observeReceipt​

type observeReceipt struct {
	ID        string `json:"id"`
	Scope     string `json:"scope"`
	LogOffset int64  `json:"log_offset"`
}

source

type apiError​

type apiError struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

source

type apiClient​

type apiClient struct {
	base  string
	token string
	http  *http.Client
}

source

method apiClient.observe​

func (c *apiClient) observe(ctx context.Context, body observeBody, capacityWait time.Duration) (observeReceipt, error)

observe sends one segment. A full backlog (429) is waited out with bounded backoff up to capacityWait, because a bulk import is exactly the caller that should retry with the same key; a conflict (409) is a document that changed under a key it had already used, which cannot happen with keys derived from the digest and is refused rather than papered over.

source

method apiClient.awaitFormation​

func (c *apiClient) awaitFormation(ctx context.Context, offset int64, wait time.Duration) (*int64, int, error)

awaitFormation polls freshness until the formed watermark reaches the offset or the wait ends, and returns the watermark and the parked count as they stood.

source

method apiClient.post​

func (c *apiClient) post(ctx context.Context, path string, body any) (int, []byte, error)

source

method apiClient.get​

func (c *apiClient) get(ctx context.Context, path string) (int, []byte, error)

source

method apiClient.do​

func (c *apiClient) do(req *http.Request) (int, []byte, error)

source