dagsync

package
v0.1.23-oss Latest Latest
Warning

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

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

README

dagsync

dagsync is a Go library designed to synchronize Directed Acyclic Graphs (DAGs) of objects to a remote service. It handles dependency resolution, concurrent uploads, retries, and maintenance of the object graph.

Overview

The core component is the DAGSync struct, which orchestrates the upload process. It maintains a graph of nodes representing objects to be synced.

  • Automated Dependency Management: Automatically discovers links between objects (e.g., a Tree object linking to Blobs or other Trees) and ensures dependencies are uploaded before the parent object.
  • Concurrency: Uploads objects in parallel using a configurable number of workers.
  • Retries: Automatically retries failed uploads with an exponential backoff.
  • Maintenance: Cleans up placeholder nodes that are no longer needed.

Usage

Initialization

Create a new DAGSync instance by providing an ABFSUpdater (for writing to the remote) and an ObjectAccessor (for reading local objects).

opts := &dagsync.DagSyncOpts{
    Updater:      myUpdater,
    Accessor:     myAccessor,
    WorkersCount: 10,
    // Optional callbacks
    OnPush: func(h entities.Hash, t string) {
        fmt.Printf("Pushed %s (%s)\n", h, t)
    },
}
d := dagsync.New(opts)
Starting

Start the synchronization processing loop in a background goroutine. This method blocks until the context is cancelled.

ctx, cancel := context.WithCancel(context.Background())
go d.Start(ctx)
Adding Requests

You can request objects to be synced using various methods. dagsync will automatically handle their dependencies.

  • PutObject(hash): Syncs a generic object.
  • PutBlob(hash): Syncs a blob object.
  • PutTree(hash): Syncs a tree object.
  • AddMapResult(from, to): Records a mapping between two objects.
  • PutRefs(server, project, refs): Updates references on a server.
  • Phony(priority, objects, callback): Phony requests do not request any data to be synced, but provide features on existing requests and query the the remote server for status. A priority can be provided which requests dagsync to prioritize the provided objects (and their dependencies) over others. Also, a callback can be provided to be notified on success or failure. The actual running of the request queries the remote server to see if the provided objects have been synced.

All these methods return a fmt.Stringer request object immediately and process the upload asynchronously. The return value is unique for an object (e.g. putting the same object again will return the same string) and can be used to query the state of a request.

Configuration

  • WorkersCount: Number of concurrent upload workers.
  • FailCooldown: Base duration to wait before retrying a failed node.
  • PlaceholderNodeTimeout: Duration to keep "placeholder" nodes (nodes created as dependencies but not yet fully known) before cleaning them up if they remain orphaned.
  • MaintenanceRunInterval: Interval at which the maintenance loop runs to clean up stale nodes.

Architecture

dagsync uses a priority queue to manage the execution order of nodes. Nodes with satisfied dependencies are moved to a QueuedState and processed by workers. If a node fails, it enters a FailedState and is rescheduled.

Documentation

Index

Constants

This section is empty.

Variables

View Source
var Debug = false

Functions

This section is empty.

Types

type ABFSUpdater

type ABFSUpdater interface {
	Headers(ctx context.Context, keys []remote.Hash) ([]entities.ObjectHeader, error)
	PutObject(ctx context.Context, key entities.Hash, reader remote.ContentReadCloser) error
	AddMapResult(ctx context.Context, from, to entities.Hash) error
	PutRefs(ctx context.Context, server string, project string, refs []entities.RefHash) error
}

type AddMapRequest

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

func (*AddMapRequest) Execute

func (r *AddMapRequest) Execute(ctx context.Context, updater ABFSUpdater, accessor ObjectAccessor) error

func (*AddMapRequest) Key

func (r *AddMapRequest) Key() entities.Hash
func (r *AddMapRequest) Links(accessor ObjectAccessor) ([]entities.Hash, error)

func (*AddMapRequest) String

func (r *AddMapRequest) String() string

func (*AddMapRequest) Type

func (r *AddMapRequest) Type() string

type Callback

type Callback func(err error)

type DAGSync

type DAGSync struct {
	MaintenanceRunInterval time.Duration
	FailCooldown           time.Duration
	PlaceholderNodeTimeout time.Duration

	MailboxDelay time.Duration
	// contains filtered or unexported fields
}

func New

func New(opts *DagSyncOpts) *DAGSync

func (*DAGSync) AddMapResult

func (d *DAGSync) AddMapResult(from, to entities.Hash) fmt.Stringer

func (*DAGSync) Dump

func (d *DAGSync) Dump(ctx context.Context, w io.Writer)

Dump writes the information of all nodes to the given writer (used for debugging purposes). It locks the global lock so it should not be used lightly.

func (*DAGSync) Phony

func (d *DAGSync) Phony(priority RequestPriority, objects []entities.Hash, callback Callback) fmt.Stringer

func (*DAGSync) PutBlob

func (d *DAGSync) PutBlob(key entities.Hash) fmt.Stringer

func (*DAGSync) PutObject

func (d *DAGSync) PutObject(key entities.Hash) fmt.Stringer

func (*DAGSync) PutRefs

func (d *DAGSync) PutRefs(server string, project string, refs []entities.RefHash, callbacks ...Callback) fmt.Stringer

func (*DAGSync) PutTree

func (d *DAGSync) PutTree(key entities.Hash) fmt.Stringer

func (*DAGSync) Send

func (d *DAGSync) Send(m Message)

func (*DAGSync) Start

func (d *DAGSync) Start(ctx context.Context)

func (*DAGSync) State

func (d *DAGSync) State(tag string) (NodeState, bool)

func (*DAGSync) Stats

func (d *DAGSync) Stats() Stats

func (*DAGSync) UpdateLogger

func (d *DAGSync) UpdateLogger(logger *log.Logger)

func (*DAGSync) Validate

func (d *DAGSync) Validate(ctx context.Context, w io.Writer, nodeStaleTimeout time.Duration)

Validate checks the internal state and prints any inconsistentcies to the given writer (used for debugging purposes). It locks the global lock so it should not be used lightly.

func (*DAGSync) Wait

func (d *DAGSync) Wait(timeoutDuration time.Duration) error

Wait() returns when dagsync is idle or the timeoutDuration is reached. It returns error if there are any broken entries after the dagsync is idle.

type DagSyncOpts

type DagSyncOpts struct {
	Updater      ABFSUpdater
	Accessor     ObjectAccessor
	WorkersCount int

	PeekOffline func(entities.Hash) (bool, bool)
	OnPush      func(entities.Hash, string)

	// If = 0, skipping is disabled.
	// If enabled, the ObjectAccessor must also be a SizeProvider.
	OfflineThreshold uint64
	Logger           *log.Logger
}

type Mailbox

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

Mailbox provides a similar interface to a channel, but with an unbound capacity.

func NewMailbox

func NewMailbox() *Mailbox

NewMailbox creates, starts, and returns a new Mailbox.

func (*Mailbox) Close

func (mb *Mailbox) Close()

Close gracefully shuts down the mailbox. Similar to golang channels, it signals there will be no more sends and Receive will be closed after all messages are consumed.

func (*Mailbox) Receive

func (mb *Mailbox) Receive() <-chan Message

Receive returns the read-only channel for consuming messages.

func (*Mailbox) Send

func (mb *Mailbox) Send(msg Message)

Send posts a message to the mailbox.

type Message

type Message interface{}

type NodeState

type NodeState int
const (
	HolderState               NodeState = iota // The node is only a placeholder for the object
	InitialLinkBuildingState                   // The node is going through the initial link generation
	FailedLinkBuildState                       // The node has failed to build the links, it will be retried
	QueuedState                                // The node is ready to be executed and is added to the queue
	BlockedState                               // The node is blocked by one of its dependencies
	RunningState                               // The node is being executed
	FixingLinkValidationState                  // The node failed to execute due to a link validation error, and it is being fixed before re-activating
	FailedState                                // The execution of the node has failed, it will be retried

)

func (NodeState) IsLinked

func (s NodeState) IsLinked() bool

func (NodeState) String

func (s NodeState) String() string

type ObjectAccessor

type ObjectAccessor interface {
	GetObject(ctx context.Context, key entities.Hash) ([]byte, error)
	// OpenBlob is required to return a ReadSizeCloser because the size is needed as a header
	// to the object. If the size isn't provided, all the bytes must be read into memory, which
	// is an issue for concurrent large blobs.
	OpenBlob(ctx context.Context, key entities.Hash) (ReadSizeCloser, error)
}

type PhonyRequest

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

func (*PhonyRequest) Execute

func (r *PhonyRequest) Execute(ctx context.Context, updater ABFSUpdater, accessor ObjectAccessor) error

func (*PhonyRequest) Key

func (r *PhonyRequest) Key() entities.Hash
func (r *PhonyRequest) Links(_ ObjectAccessor) ([]entities.Hash, error)

func (*PhonyRequest) String

func (r *PhonyRequest) String() string

func (*PhonyRequest) Type

func (r *PhonyRequest) Type() string

type PriorityQueue

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

PriorityQueue orders by greater priority value then earlier time.

func NewPriorityQueue

func NewPriorityQueue() *PriorityQueue

func (*PriorityQueue) Len

func (q *PriorityQueue) Len() int

Len returns the number of items safely.

func (*PriorityQueue) Peek

func (q *PriorityQueue) Peek() *node

func (*PriorityQueue) Push

func (q *PriorityQueue) Push(item *node)

func (*PriorityQueue) Remove

func (q *PriorityQueue) Remove(n *node)

func (*PriorityQueue) Update

func (p *PriorityQueue) Update(n *node, priority RequestPriority)

type PutObjectRequest

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

func (*PutObjectRequest) Execute

func (r *PutObjectRequest) Execute(ctx context.Context, updater ABFSUpdater, accessor ObjectAccessor) error

func (*PutObjectRequest) IsBlob

func (r *PutObjectRequest) IsBlob() bool

func (*PutObjectRequest) Key

func (r *PutObjectRequest) Key() entities.Hash
func (r *PutObjectRequest) Links(accessor ObjectAccessor) ([]entities.Hash, error)

func (*PutObjectRequest) String

func (r *PutObjectRequest) String() string

func (*PutObjectRequest) Type

func (r *PutObjectRequest) Type() string

type PutRefsRequest

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

func (*PutRefsRequest) Execute

func (r *PutRefsRequest) Execute(ctx context.Context, updater ABFSUpdater, accessor ObjectAccessor) error

func (*PutRefsRequest) Key

func (r *PutRefsRequest) Key() entities.Hash
func (r *PutRefsRequest) Links(accessor ObjectAccessor) ([]entities.Hash, error)

func (*PutRefsRequest) String

func (r *PutRefsRequest) String() string

func (*PutRefsRequest) Type

func (r *PutRefsRequest) Type() string

type QueuePriority

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

type ReadSizeCloser

type ReadSizeCloser interface {
	io.ReadCloser
	entities.Sizer
}

type Request

type Request interface {
	Links(accessor ObjectAccessor) ([]entities.Hash, error)
	Execute(ctx context.Context, updater ABFSUpdater, accessor ObjectAccessor) error
	Type() string
	String() string
	Key() entities.Hash
}

func NewObjectRequest

func NewObjectRequest(hash entities.Hash) Request

NewObjectRequest creates an Object request based on the provided type

type RequestPriority

type RequestPriority float64

type SizeProvider

type SizeProvider interface {
	ObjectSize(ctx context.Context, objectHash entities.Hash) (int64, error)
	BlobSize(ctx context.Context, objectHash entities.Hash) (int64, error)
}

type Stats

type Stats struct {
	States         map[string]map[string]int64
	BrokenLinks    int64
	MissingObjects int64
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL