worker

Documentation

Overview

* SPDX-FileCopyrightText: bonsai contributors * SPDX-License-Identifier: Apache-2.0 * * Local mutation apply path. Replaces upstream's RaftProposal-based * MutateOverNetwork — there is no Raft, no group routing. Each mutation runs * against the local Badger directly via posting.Txn / List.AddMutationWithIndex. * * Ported from priorart/dgraph/worker/mutation.go::runMutation + * MutateOverNetwork, with all cluster forwarding removed.

* SPDX-FileCopyrightText: bonsai contributors * SPDX-License-Identifier: Apache-2.0 * * Single-node minimal worker package. * * Upstream Dgraph's `worker` package was the home of the cluster machinery: * Raft, group routing, predicate-move choreography, the inter-alpha gRPC * services, the distributed Oracle, and the local query/mutation execution * engine all lived together. In bonsai the cluster pieces are gone, so * `worker` shrinks to a thin façade that: * * - holds the Badger handle (`pstore`) * - holds the per-process UID counter that replaces upstream's xidmap * - exposes the API surface the `posting` and `query` packages still call * (`MutateOverNetwork`, `ProcessTaskOverNetwork`, `SortOverNetwork`, * `AssignUidsOverNetwork`, `GetSchemaOverNetwork`, `GetTypes`, * `Init`, `StartRaftNodes`, `MaxLeaseId`, `LimitDefaults`, * `ErrNonExistentTabletMessage`) * * The "OverNetwork" mutation/query/sort entry points currently return * ErrNotImplemented — they compile but do not yet execute. The bonsai demo * library (pkg/bonsai) drives Badger and posting directly for now.

Index

• Constants
• Variables
• func ApplyInitialSchema(ns, ts uint64) error
• func AssignUidsOverNetwork(_ context.Context, num *pb.Num) (*pb.AssignedIds, error)
• func BlockingStop()
• func CurrentTs() uint64
• func GetSchemaOverNetwork(_ context.Context, req *pb.SchemaRequest) ([]*pb.SchemaNode, error)
• func GetTypes(_ context.Context, req *pb.SchemaRequest) ([]*pb.TypeUpdate, error)
• func Init(ps *badger.DB)
• func MaxLeaseId() uint64
• func MutateOverNetwork(ctx context.Context, m *pb.Mutations) (*api.TxnContext, error)
• func NextTs() uint64
• func ProcessTaskOverNetwork(ctx context.Context, q *pb.Query) (*pb.Result, error)
• func Pstore() *badger.DB
• func SeedLocalTs(ts uint64)
• func SetMaxUID(uid uint64)
• func SortOverNetwork(ctx context.Context, q *pb.SortMessage) (*pb.SortResult, error)
• func StartRaftNodes(_ string)
• func ValidateAndConvert(edge *pb.DirectedEdge, su *pb.SchemaUpdate) error
• type FuncType
• type Options

Constants

const (
	// UseTxnCache indicates the transaction cache should be used.
	UseTxnCache = iota
	// NoCache indicates no caches should be used.
	NoCache
)

const ErrNonExistentTabletMessage = "Requested predicate is not being served by this server."

ErrNonExistentTabletMessage matches upstream's error string so that callers in `query/query.go` that string-compare against it still work.

const LimitDefaults = `mutations=allow; query-edge=1000000; normalize-node=10000; ` +
	`mutations-nquad=1000000; disallow-drop=false; query-timeout=0ms; txn-abort-after=5m; ` +
	`max-retries=10; max-pending-queries=10000; shared-instance=false; type-filter-uid-limit=10`

LimitDefaults mirrors upstream's `--limit` superflag string. Some flags (e.g. `mutations`) are read by code paths that survive in bonsai.

const MaxLeaseUid = uint64(1) << 62

MaxLeaseUid is the largest UID we will ever hand out. Matches upstream constant so query/mutation.go's overflow check still triggers correctly.

Variables

var ErrNotImplemented = errors.New("bonsai: query/mutation engine not yet wired up (P1 work-in-progress)")

ErrNotImplemented is the placeholder error returned by query/mutation entry points that have not yet been ported to the single-node engine.

Functions

func ApplyInitialSchema

func ApplyInitialSchema(ns, ts uint64) error

ApplyInitialSchema persists the reserved (`dgraph.*`) schema and type definitions for the given namespace. Called by pkg/bonsai.Open for the root namespace and by CreateNamespace for new tenants.

func AssignUidsOverNetwork

func AssignUidsOverNetwork(_ context.Context, num *pb.Num) (*pb.AssignedIds, error)

AssignUidsOverNetwork hands out a contiguous block of fresh UIDs. Replaces upstream's Zero-leased xidmap with a local atomic counter.

func BlockingStop

func BlockingStop()

BlockingStop tears down worker state. Called by pkg/bonsai.Close.

func CurrentTs

func CurrentTs() uint64

CurrentTs returns the current high-water without advancing it.

func GetSchemaOverNetwork

func GetSchemaOverNetwork(_ context.Context, req *pb.SchemaRequest) ([]*pb.SchemaNode, error)

GetSchemaOverNetwork looks up schema definitions from the local schema state. Reads the in-memory schema cache populated by Alter.

func GetTypes

func GetTypes(_ context.Context, req *pb.SchemaRequest) ([]*pb.TypeUpdate, error)

GetTypes fetches type definitions by name. Reads the local schema state.

func Init

func Init(ps *badger.DB)

Init wires the Badger handle into the worker package. Called by pkg/bonsai.Open.

func MaxLeaseId

func MaxLeaseId() uint64

MaxLeaseId returns the highest UID currently assigned.

func MutateOverNetwork

func MutateOverNetwork(ctx context.Context, m *pb.Mutations) (*api.TxnContext, error)

MutateOverNetwork applies a mutation to the local store. Replaces the upstream cluster-routed implementation with a serialised local apply.

The flow is:

1. Allocate startTs.
2. Apply schema/type updates.
3. Walk edges; runMutation each one through a fresh posting.Txn.
4. Allocate commitTs and CommitToDisk via a TxnWriter.
5. Tell the posting Oracle the txn committed so subsequent reads are visible at >= commitTs.

func NextTs

func NextTs() uint64

NextTs is the public version of nextLocalTs, exposed so pkg/bonsai can share the same atomic counter for its non-Mutate write paths (DB.Set, DB.Alter, DB.AssignUid persistence). Keeping a single counter avoids the dual-counter bug where reads at d.tsCount blocked forever in Oracle.WaitForTs because worker had advanced past it.

func ProcessTaskOverNetwork

func ProcessTaskOverNetwork(ctx context.Context, q *pb.Query) (*pb.Result, error)

ProcessTaskOverNetwork is the entry point query/ uses to fetch UIDs/values for a single predicate. In bonsai there are no remote groups, so this is just a thin wrapper around the local processTask.

func Pstore

func Pstore() *badger.DB

Pstore returns the Badger handle. Used by posting/.

func SeedLocalTs

func SeedLocalTs(ts uint64)

SeedLocalTs is called by pkg/bonsai.Open to seed the local timestamp counter from the recovered Badger MaxVersion. Must be called before any mutation is processed.

func SetMaxUID

func SetMaxUID(uid uint64)

SetMaxUID bumps the UID counter, used by Restore and by AssignUids.

func SortOverNetwork

func SortOverNetwork(ctx context.Context, q *pb.SortMessage) (*pb.SortResult, error)

SortOverNetwork sorts a UID matrix locally. bonsai has no remote groups, so this is a thin wrapper around processSort.

func StartRaftNodes

func StartRaftNodes(_ string)

StartRaftNodes is a no-op in bonsai; cluster bootstrap is gone. Retained as a symbol because tests under query/ still call it.

func ValidateAndConvert

func ValidateAndConvert(edge *pb.DirectedEdge, su *pb.SchemaUpdate) error

ValidateAndConvert is a near-verbatim port of the upstream helper. The only behavioural changes: vector-keyword check uses the bonsai hnsw constant reference (same value); the ACL `dgraph.rule.permission` branch is dropped because bonsai has no ACL.

Types

type FuncType

type FuncType int

FuncType represents the type of a query function (aggregation, has, etc).

type Options

type Options struct {
	PostingDir         string
	WALDir             string
	MyAddr             string
	HmacSecret         []byte
	LudicrousMode      bool
	TypeFilterUidLimit uint64
}

Options is the runtime configuration for the worker subsystem. ACL, encryption, vault, and CDC fields are gone — bonsai is local-only.

var Config Options

Config is the global, mutable worker configuration. Surviving fields are the ones still referenced from the schema/posting layers.