casync

README

Go casync

This project doesn't seek to be a full reimplementation of upstream casync, but is a collection of tools that utilize certain aspects of casync for very specific use cases.

Goals And Non-Goals

Among the distinguishing factors:

• Where the upstream command has chosen to optimize for storage efficiency (f/e, being able to use local files as "seeds", building temporary indexes into them), this command chooses to optimize for runtime performance (maintaining a local explicit chunk store, avoiding the need to reindex) at cost to storage efficiency.

• Where the upstream command has chosen to take full advantage of Linux platform features, this client chooses to implement a minimum featureset and, while high-value platform-specific features (such as support for btrfs reflinks into a decompressed local chunk cache) might be added in the future, the ability to build without them on other platforms will be maintained.

• Supporting the .catar archive format (whether or not packaged in a .caidx) is not presently a goal.

• SHA512/256 is currently the only supported hash function.

• Only chunk store using zstd compression are supported at this point.

Tools

desync

Basic client to read casync blob index files (caibx) and reassemble a blob from chunks read from a chunk store (local or remove via SSH).

Options

• -s <store> Location of the chunk store, can be local directory or a URL like ssh://hostname/path/to/store
• -c <store> Location of a local chunk store to be used as cache. Needs to be writable.
• -n <int> Number of concurrent download jobs and ssh sessions to the chunk store.

Environment variables

• CASYNC_SSH_PATH overrides the default "ssh" with a command to run when connecting to the remove chunk store
• CASYNC_REMOTE_PATH defines the command to run on the chunk store when using SSH, default "casync"

Caching

The -c <store> option can be used to either specify an existing local store to act as cache or to build a new one. Whenever a cache is requested, it is first looked up in the local cache before routing the request to the main (likely remote store). Any chunks downloaded from the main store are added to the local store (cache). In addition, when a chunk is read from the cache, mtime of the chunk is updated to allow for basic garbage collection based on file age. The cache directory as well as the chunks in it are expected to be writable. If the cache contains an invalid chunk (checksum does not match the chunk ID), blob assembly will fail. Invalid chunks are not skipped or removed from the cache automatically.

Examples:

Re-assemble somefile.tar using a remote chunk store and a blob index file.

desync -s ssh://192.168.1.1/path/to/casync.store/ -c /tmp/store somefile.tar.caibx somefile.tar

TODOs

• Write tests
• Pre-allocate the output file to avoid fragmentation
• Check output file size, compare to expected size
• Support retrieval of index files from the chunk store
• Allow on-disk chunk cache to optionally be stored uncompressed, such that blocks can be directly reflinked (rather than copied) into files, when on a platform and filesystem where reflink support is available.
• When using the remote store, multiple SSH sessions and csync processes are started, there's nothing to stop them yet (relies on process shutdown/cleanup)
• Code cleanup and reorg

Documentation

Index

• Constants
• func DecompressInto(w io.Writer, b []byte) (int64, error)
• type BlobIndexChunk
• type Cache
  • func NewCache(s Store, l LocalStore) Cache
  • func (c Cache) GetChunk(id ChunkID) ([]byte, error)
• type Caibx
  • func CaibxFromReader(r io.Reader) (c Caibx, err error)
• type ChunkID
  • func ChunkIDFromSlice(b []byte) (ChunkID, error)
  • func ChunkIDFromString(id string) (ChunkID, error)
  • func (c ChunkID) String() string
• type ChunkMissing
  • func (e ChunkMissing) Error() string
• type Header
• type Index
• type LocalStore
  • func NewLocalStore(dir string) (LocalStore, error)
  • func (s LocalStore) GetChunk(id ChunkID) ([]byte, error)
  • func (s LocalStore) StoreChunk(id ChunkID, b []byte) error
• type Message
• type RemoteSSH
  • func NewRemoteSSHStore(location *url.URL, n int) (*RemoteSSH, error)
  • func (s *RemoteSSH) GetChunk(id ChunkID) ([]byte, error)
• type Session
  • func StartSession(host string, path string) (Session, error)
  • func (s Session) ReadMessage() (Message, error)
  • func (s Session) RecvHello() (uint64, error)
  • func (s Session) RequestChunk(id ChunkID) ([]byte, error)
  • func (s Session) SendHello(flags uint64) error
  • func (s Session) SendProtocolRequest(id ChunkID, flags uint64) error
  • func (s Session) WriteMessage(m Message) error
• type Store

Constants

const (
	// Format identifiers
	CaFormatIndex  = 0x96824D9C7B129FF9
	CaFormatHeader = 0xE75B9E112F17417D

	// Protocol message types
	CaProtocolHello      = 0x3c71d0948ca5fbee
	CaProtocolIndex      = 0xb32a91dd2b3e27f8
	CaProtocolIndexEOF   = 0x4f0932f1043718f5
	CaProtocolArchive    = 0x95d6428a69eddcc5
	CaProtocolArchiveEOF = 0x450bef663f24cbad
	CaProtocolRequest    = 0x8ab427e0f89d9210
	CaProtocolChunk      = 0x5213dd180a84bc8c
	CaProtocolMissing    = 0xd010f9fac82b7b6c
	CaProtocolGoodbye    = 0xad205dbf1a3686c3
	CaProtocolAbort      = 0xe7d9136b7efea352

	// Provided services
	CaProtocolReadableStore   = 0x1
	CaProtocolWritableStore   = 0x2
	CaProtocolReadableIndex   = 0x4
	CaProtocolWritableIndex   = 0x8
	CaProtocolReadableArchive = 0x10
	CaProtocolWritableArchive = 0x20

	// Wanted services
	CaProtocolPullChunks      = 0x40
	CaProtocolPullIndex       = 0x80
	CaProtocolPullArchive     = 0x100
	CaProtocolPushChunks      = 0x200
	CaProtocolPushIndex       = 0x400
	CaProtocolPushIndexChunks = 0x800
	CaProtocolPushArchive     = 0x1000

	// Protocol request flags
	CaProtocolRequestHighPriority = 1
)

Functions

func DecompressInto

func DecompressInto(w io.Writer, b []byte) (int64, error)

DecompressInto takes a raw (compressed) chunk as returned by the store and unpacks it into a provided writer. Returns the number of bytes written.

Types

type BlobIndexChunk

type BlobIndexChunk struct {
	Start uint64
	Size  uint64
	ID    ChunkID
}

BlobIndexChunk is a table entry in a caibx file containing the chunk ID (SHA256) as well as the offset within the blob after appending this chunk

type Cache

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

Cache is used to connect a (typically remote) store with a local store which functions as disk cache. Any request to the cache for a chunk will first be routed to the local store, and if that fails to the slower remote store. Any chunks retrieved from the remote store will be stored in the local one.

func NewCache

func NewCache(s Store, l LocalStore) Cache

NewCache returns a cache router that uses a local store as cache before accessing a (supposedly slower) remote one.

func (Cache) GetChunk

func (c Cache) GetChunk(id ChunkID) ([]byte, error)

GetChunk first asks the local store for the chunk and then the remote one. If we get a chunk from the remote, it's stored locally too.

type Caibx

type Caibx struct {
	Index  Index
	Header Header
	Chunks []BlobIndexChunk
}

Caibx represents the content of a caibx file

func CaibxFromReader

func CaibxFromReader(r io.Reader) (c Caibx, err error)

CaibxFromReader parses a caibx structure (from a reader) and returns a populated Caibx object

type ChunkID

type ChunkID [32]byte

ChunkID is the SHA512/256 in binary encoding

func ChunkIDFromSlice

func ChunkIDFromSlice(b []byte) (ChunkID, error)

ChunkIDFromSlice converts a SHA512/256 encoded as byte slice into a ChunkID. It's expected the slice is of the correct length

func ChunkIDFromString

func ChunkIDFromString(id string) (ChunkID, error)

ChunkIDFromString converts a SHA512/56 encoded as string into a ChunkID

func (ChunkID) String

func (c ChunkID) String() string

type ChunkMissing

type ChunkMissing struct {
	ID ChunkID
}

ChunkMissing is returned by a store that can't find a requested chunk

func (ChunkMissing) Error

func (e ChunkMissing) Error() string

type Header

type Header struct {
	Size uint64
	Type uint64
}

Header follows the Index in the caibx file

type Index

type Index struct {
	Size         uint64
	Type         uint64
	Flags        uint64
	ChunkSizeMin uint64
	ChunkSizeAvg uint64
	ChunkSizeMax uint64
}

Index at the start of the caibx file

type LocalStore

type LocalStore struct {
	Base string

	// When accessing chunks, should mtime be updated? Useful when this is
	// a cache. Old chunks can be identified and removed from the store that way
	UpdateTimes bool
}

LocalStore casync store

func NewLocalStore

func NewLocalStore(dir string) (LocalStore, error)

NewLocalStore creates an instance of a local castore, it only checks presence of the store

func (LocalStore) GetChunk

func (s LocalStore) GetChunk(id ChunkID) ([]byte, error)

GetChunk reads and returns one (compressed!) chunk from the store

func (LocalStore) StoreChunk

func (s LocalStore) StoreChunk(id ChunkID, b []byte) error

StoreChunk adds a new chunk to the store

type Message

type Message struct {
	Type uint64
	Body []byte
}

type RemoteSSH

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

RemoteSSH is a remote casync store accessed via SSH. Supports running multiple sessions to improve throughput.

func NewRemoteSSHStore

func NewRemoteSSHStore(location *url.URL, n int) (*RemoteSSH, error)

NewRemoteSSHStore establishes up to n connections with a casync chunk server

func (*RemoteSSH) GetChunk

func (s *RemoteSSH) GetChunk(id ChunkID) ([]byte, error)

GetChunk requests a chunk from the server and returns a (compressed) one. It uses any of the n sessions this store maintains in its pool. Blocks until one session becomes available

type Session

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

Session represents one instance of a local SSH client and a remote casync store server. TODO: There's currently no way to stop a session, we just rely on the process terminating and cleaning up everything.

func StartSession

func StartSession(host string, path string) (Session, error)

StartSession initiates a connection to the remote store server using the value in CASYNC_SSH_PATH (default "ssh"), and executes the command in CASYNC_REMOTE_PATH (default "casync"). It then performs the HELLO handshake to initialze the connection before returning a Session object that can then be used to request chunks.

func (Session) ReadMessage

func (s Session) ReadMessage() (Message, error)

ReadMessage reads a generic message from the server, verifies the length, extracts the type and returns the message body as byte slice

func (Session) RecvHello

func (s Session) RecvHello() (uint64, error)

RecvHello waits for the server to send a HELLO, fails if anything else is received. Returns the flags provided by the server.

func (Session) RequestChunk

func (s Session) RequestChunk(id ChunkID) ([]byte, error)

RequestChunk sends a request for a specific chunk to the server, waits for the response and returns the bytes in the chunk. Returns an error if the server reports the chunk as missing

func (Session) SendHello

func (s Session) SendHello(flags uint64) error

SendHello sends a HELLO message to the server, with the flags signaling which service is being requested from it.

func (Session) SendProtocolRequest

func (s Session) SendProtocolRequest(id ChunkID, flags uint64) error

func (Session) WriteMessage

func (s Session) WriteMessage(m Message) error

WriteMessage sends a generic message to the server

type Store

type Store interface {
	GetChunk(id ChunkID) ([]byte, error)
}