Skip to content

Add NATS replica client #596

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
delaneyj wants to merge 3 commits into from
Closed

Add NATS replica client #596

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
008dbab
initial commit
delaneyj Jul 31, 2024
eff24b7
add nats to cli
delaneyj Aug 22, 2024
f9898e4
fixes from PR review
delaneyj Aug 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
.DS_Store
/dist
.vscode
/dist
39 changes: 28 additions & 11 deletions file/replica_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string)
func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (litestream.SnapshotIterator, error) {
dir, err := c.SnapshotsDir(generation)
if err != nil {
return nil, fmt.Errorf("cannot determine snapshots path: %w", err)
return nil, litestream.ErrSnapshotPathNoGeneration
}

f, err := os.Open(dir)
Expand Down Expand Up @@ -186,7 +186,7 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (lites
func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, index int, rd io.Reader) (info litestream.SnapshotInfo, err error) {
filename, err := c.SnapshotPath(generation, index)
if err != nil {
return info, fmt.Errorf("cannot determine snapshot path: %w", err)
return info, litestream.ErrSnapshotPathNoGeneration
}

var fileInfo, dirInfo os.FileInfo
Expand Down Expand Up @@ -239,16 +239,23 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, index int) (io.ReadCloser, error) {
filename, err := c.SnapshotPath(generation, index)
if err != nil {
return nil, fmt.Errorf("cannot determine snapshot path: %w", err)
return nil, litestream.ErrSnapshotDoesNotExist
}
return os.Open(filename)
f, err := os.Open(filename)
if err != nil {
if os.IsNotExist(err) {
return nil, litestream.ErrSnapshotDoesNotExist
}
return nil, err
}
return f, nil
}

// DeleteSnapshot deletes a snapshot with the given generation & index.
func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, index int) error {
filename, err := c.SnapshotPath(generation, index)
if err != nil {
return fmt.Errorf("cannot determine snapshot path: %w", err)
return litestream.ErrSnapshotDoesNotExist
}
if err := os.Remove(filename); err != nil && !os.IsNotExist(err) {
return err
Expand All @@ -260,19 +267,22 @@ func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, i
func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (litestream.WALSegmentIterator, error) {
dir, err := c.WALDir(generation)
if err != nil {
return nil, fmt.Errorf("cannot determine wal path: %w", err)
return nil, litestream.ErrWALPathNoGeneration
}

f, err := os.Open(dir)
if os.IsNotExist(err) {
return litestream.NewWALSegmentInfoSliceIterator(nil), nil
} else if err != nil {
return nil, err
return nil, litestream.ErrWALPathNoGeneration
}
defer f.Close()

fis, err := f.Readdir(-1)
if err != nil {
if err == os.ErrNotExist {
return nil, litestream.ErrWALPathNoGeneration
}
return nil, err
}

Expand Down Expand Up @@ -301,7 +311,7 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (lit
func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos, rd io.Reader) (info litestream.WALSegmentInfo, err error) {
filename, err := c.WALSegmentPath(pos.Generation, pos.Index, pos.Offset)
if err != nil {
return info, fmt.Errorf("cannot determine wal segment path: %w", err)
return info, litestream.ErrWALPathNoGeneration
}

var fileInfo, dirInfo os.FileInfo
Expand Down Expand Up @@ -355,17 +365,24 @@ func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos,
func (c *ReplicaClient) WALSegmentReader(ctx context.Context, pos litestream.Pos) (io.ReadCloser, error) {
filename, err := c.WALSegmentPath(pos.Generation, pos.Index, pos.Offset)
if err != nil {
return nil, fmt.Errorf("cannot determine wal segment path: %w", err)
return nil, litestream.ErrWALSegmentPathNoGeneration
}
f, err := os.Open(filename)
if err != nil {
if os.IsNotExist(err) {
return nil, litestream.ErrSnapshotDoesNotExist
}
return nil, err
}
return os.Open(filename)
return f, nil
}

// DeleteWALSegments deletes WAL segments at the given positions.
func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Pos) error {
for _, pos := range a {
filename, err := c.WALSegmentPath(pos.Generation, pos.Index, pos.Offset)
if err != nil {
return fmt.Errorf("cannot determine wal segment path: %w", err)
return litestream.ErrWALSegmentPathNoGeneration
}
if err := os.Remove(filename); err != nil && !os.IsNotExist(err) {
return err
Expand Down
17 changes: 10 additions & 7 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,14 @@ require (
github.com/aws/aws-sdk-go v1.49.5
github.com/mattn/go-shellwords v1.0.12
github.com/mattn/go-sqlite3 v1.14.19
github.com/nats-io/nats.go v1.36.0
github.com/pierrec/lz4/v4 v4.1.19
github.com/pkg/sftp v1.13.6
github.com/prometheus/client_golang v1.17.0
golang.org/x/crypto v0.17.0
golang.org/x/sync v0.5.0
golang.org/x/sys v0.15.0
github.com/valyala/bytebufferpool v1.0.0
golang.org/x/crypto v0.25.0
golang.org/x/sync v0.7.0
golang.org/x/sys v0.22.0
google.golang.org/api v0.154.0
gopkg.in/yaml.v2 v2.4.0
)
Expand All @@ -32,16 +34,17 @@ require (
github.com/go-logr/stdr v1.2.2 // indirect
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
github.com/golang/protobuf v1.5.3 // indirect
github.com/google/go-cmp v0.6.0 // indirect
github.com/google/s2a-go v0.1.7 // indirect
github.com/google/uuid v1.5.0 // indirect
github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
github.com/googleapis/gax-go/v2 v2.12.0 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
github.com/klauspost/compress v1.17.9 // indirect
github.com/kr/fs v0.1.0 // indirect
github.com/mattn/go-ieproxy v0.0.11 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
github.com/nats-io/nkeys v0.4.7 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
github.com/prometheus/client_model v0.5.0 // indirect
github.com/prometheus/common v0.45.0 // indirect
github.com/prometheus/procfs v0.12.0 // indirect
Expand All @@ -51,9 +54,9 @@ require (
go.opentelemetry.io/otel v1.21.0 // indirect
go.opentelemetry.io/otel/metric v1.21.0 // indirect
go.opentelemetry.io/otel/trace v1.21.0 // indirect
golang.org/x/net v0.19.0 // indirect
golang.org/x/net v0.21.0 // indirect
golang.org/x/oauth2 v0.15.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/text v0.16.0 // indirect
golang.org/x/time v0.5.0 // indirect
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
google.golang.org/appengine v1.6.8 // indirect
Expand Down
Loading