feat(remote): push & pull logs on publish & add

api/remote_test.go

@@ -25,29 +25,4 @@ func TestRemoteClientHandlers(t *testing.T) {
  {"GET", "/fetch/me/cities", nil},
  }
  runHandlerTestCases(t, "fetch", h.NewFetchHandler("/fetch"), fetchCases, true)
-
- // node, teardown := newTestNode(t)
- // defer teardown()
-
- // inst := newTestInstanceWithProfileFromNode(node)
-
- // // Set a seed so that the sessionID is deterministic
- // rand.Seed(1234)
-
- // testCases := []handlerTestCase{
- // {"POST", "/", mustFile(t, "testdata/postRemoteRequest.json")},
- // }
-
- // cfg, _ := testConfigAndSetter()
- // // testReceivers := dsync.NewTestReceivers()
-
- // // Reject all dag.Info's
- // cfg.API.RemoteAcceptSizeMax = 0
- // rh := NewRemoteHandlers(inst)
- // runHandlerTestCases(t, "remote reject", rh.ReceiveHandler, testCases, true)
-
- // // Accept all dag.Info's
- // cfg.API.RemoteAcceptSizeMax = -1
- // rh = NewRemoteHandlers(inst)
- // runHandlerTestCases(t, "remote accept", rh.ReceiveHandler, testCases, true)
 }

lib/datasets.go

@@ -652,6 +652,14 @@ func (r *DatasetRequests) Add(p *AddParams, res *repo.DatasetRef) (err error) {
  p.RemoteAddr = r.inst.cfg.Registry.Location
  }
 
+ // TODO (b5) - we're early in log syncronization days. This is going to fail a bunch
+ // while we work to upgrade the stack. Long term we may want to consider a mechanism
+ // for allowing partial completion where only one of logs or dataset pulling works
+ // by doing both in parallel and reporting issues on both
+ if pullLogsErr := r.inst.RemoteClient().PullLogs(ctx, repo.ConvertToDsref(ref), p.RemoteAddr); pullLogsErr != nil {
+ log.Errorf("pulling logs: %s", pullLogsErr)
+ }
+
  if err = actions.AddDataset(ctx, r.node, r.inst.RemoteClient(), p.RemoteAddr, &ref); err != nil {
  return err
  }

lib/remote.go

@@ -60,8 +60,7 @@ func (r *RemoteMethods) Fetch(p *FetchParams, res *repo.DatasetRef) error {
 
  // TODO (b5) - need contexts yo
  ctx := context.TODO()
-
- return r.cli.Fetch(ctx, repo.ConvertToDsref(ref), addr)
+ return r.cli.PullLogs(ctx, repo.ConvertToDsref(ref), addr)
 }
 
 // PublicationParams encapsulates parmeters for dataset publication
@@ -96,6 +95,14 @@ func (r *RemoteMethods) Publish(p *PublicationParams, res *repo.DatasetRef) erro
  // TODO (b5) - need contexts yo
  ctx := context.TODO()
 
+ // TODO (b5) - we're early in log syncronization days. This is going to fail a bunch
+ // while we work to upgrade the stack. Long term we may want to consider a mechanism
+ // for allowing partial completion where only one of logs or dataset pushing works
+ // by doing both in parallel and reporting issues on both
+ if pushLogsErr := r.cli.PushLogs(ctx, repo.ConvertToDsref(ref), addr); pushLogsErr != nil {
+ log.Errorf("pushing logs: %s", pushLogsErr)
+ }
+
  if err = r.cli.PushDataset(ctx, ref, addr); err != nil {
  return err
  }
@@ -128,6 +135,14 @@ func (r *RemoteMethods) Unpublish(p *PublicationParams, res *repo.DatasetRef) er
  // TODO (b5) - need contexts yo
  ctx := context.TODO()
 
+ // TODO (b5) - we're early in log syncronization days. This is going to fail a bunch
+ // while we work to upgrade the stack. Long term we may want to consider a mechanism
+ // for allowing partial completion where only one of logs or dataset pushing works
+ // by doing both in parallel and reporting issues on both
+ if removeLogsErr := r.cli.RemoveLogs(ctx, repo.ConvertToDsref(ref), addr); removeLogsErr != nil {
+ log.Error("removing logs: %s", removeLogsErr.Error())
+ }
+
  if err := r.cli.RemoveDataset(ctx, ref, addr); err != nil {
  return err
  }

logbook/logsync/logsync.go

@@ -29,20 +29,6 @@ type Logsync struct {
  didReceive Hook
 }
 
-// remove is an internal interface for methods available on foreign logbooks
-// the logsync struct contains the canonical implementation of a remote
-// interface. network clients wrap the remote interface with network behaviours,
-// using Logsync methods to do the "real work" and echoing that back across the
-// client protocol
-type remote interface {
- put(ctx context.Context, author log.Author, r io.Reader) error
- get(ctx context.Context, author log.Author, ref dsref.Ref) (sender log.Author, data io.Reader, err error)
- del(ctx context.Context, author log.Author, ref dsref.Ref) error
-}
-
-// assert at compile-time that Logsync is a remote
-var _ remote = (*Logsync)(nil)
-
 // Options encapsulates runtime configuration for a remote
 type Options struct {
  // ReceiveCheck is called before accepting a log, returning an error from this
@@ -81,63 +67,20 @@ type Hook func(ctx context.Context, author log.Author, path string) error
 
 // Author is the local author of lsync's logbook
 func (lsync *Logsync) Author() log.Author {
- return lsync.book.Author()
-}
-
-func (lsync *Logsync) put(ctx context.Context, author log.Author, r io.Reader) error {
- if lsync == nil {
- return ErrNoLogsync
- }
-
- if lsync.receiveCheck != nil {
- // TODO (b5) - need to populate path
- if err := lsync.receiveCheck(ctx, author, ""); err != nil {
- return err
- }
- }
-
- data, err := ioutil.ReadAll(r)
- if err != nil {
- return err
- }
-
- if err := lsync.book.MergeLogBytes(ctx, author, data); err != nil {
- return err
- }
-
- if lsync.didReceive != nil {
- // TODO (b5) - need to populate path
- if err := lsync.didReceive(ctx, author, ""); err != nil {
- return err
- }
- }
- return nil
-}
-
-func (lsync *Logsync) get(ctx context.Context, author log.Author, ref dsref.Ref) (log.Author, io.Reader, error) {
  if lsync == nil {
- return nil, nil, ErrNoLogsync
- }
-
- data, err := lsync.book.LogBytes(ref)
- if err != nil {
- return lsync.Author(), nil, err
+ return nil
  }
- return lsync.Author(), bytes.NewReader(data), nil
+ return lsync.book.Author()
 }
 
-func (lsync *Logsync) del(ctx context.Context, sender log.Author, ref dsref.Ref) error {
+// NewPush prepares a Push from the local logsync to a remote destination
+// doing a push places a local log on the remote
+func (lsync *Logsync) NewPush(ref dsref.Ref, remoteAddr string) (*Push, error) {
  if lsync == nil {
- return ErrNoLogsync
+ return nil, ErrNoLogsync
  }
 
- return lsync.book.RemoveLog(ctx, sender, ref)
-}
-
-// NewPush prepares a Push from the local logsync to a remote destination
-// doing a push places a local log on the remote
-func (lsync *Logsync) NewPush(ref dsref.Ref, remote string) (*Push, error) {
- rem, err := lsync.getRemote(remote)
+ rem, err := lsync.remoteClient(remoteAddr)
  if err != nil {
  return nil, err
  }
@@ -151,8 +94,12 @@ func (lsync *Logsync) NewPush(ref dsref.Ref, remote string) (*Push, error) {
 
 // NewPull creates a Pull from the local logsync to a remote destination
 // doing a pull fetches a log from the remote to the local logbook
-func (lsync *Logsync) NewPull(ref dsref.Ref, remote string) (*Pull, error) {
- rem, err := lsync.getRemote(remote)
+func (lsync *Logsync) NewPull(ref dsref.Ref, remoteAddr string) (*Pull, error) {
+ if lsync == nil {
+ return nil, ErrNoLogsync
+ }
+
+ rem, err := lsync.remoteClient(remoteAddr)
  if err != nil {
  return nil, err
  }
@@ -165,16 +112,20 @@ func (lsync *Logsync) NewPull(ref dsref.Ref, remote string) (*Pull, error) {
 }
 
 // DoRemove asks a remote to remove a log
-func (lsync *Logsync) DoRemove(ctx context.Context, ref dsref.Ref, remote string) error {
- rem, err := lsync.getRemote(remote)
+func (lsync *Logsync) DoRemove(ctx context.Context, ref dsref.Ref, remoteAddr string) error {
+ if lsync == nil {
+ return ErrNoLogsync
+ }
+
+ rem, err := lsync.remoteClient(remoteAddr)
  if err != nil {
  return err
  }
 
  return rem.del(ctx, lsync.Author(), ref)
 }
 
-func (lsync *Logsync) getRemote(remoteAddr string) (rem remote, err error) {
+func (lsync *Logsync) remoteClient(remoteAddr string) (rem remote, err error) {
  // if a valid base58 peerID is passed, we're doing a p2p dsync
  if id, err := peer.IDB58Decode(remoteAddr); err == nil {
  if lsync.p2pHandler == nil {
@@ -190,6 +141,70 @@ func (lsync *Logsync) getRemote(remoteAddr string) (rem remote, err error) {
  return rem, nil
 }
 
+// remove is an internal interface for methods available on foreign logbooks
+// the logsync struct contains the canonical implementation of a remote
+// interface. network clients wrap the remote interface with network behaviours,
+// using Logsync methods to do the "real work" and echoing that back across the
+// client protocol
+type remote interface {
+ put(ctx context.Context, author log.Author, r io.Reader) error
+ get(ctx context.Context, author log.Author, ref dsref.Ref) (sender log.Author, data io.Reader, err error)
+ del(ctx context.Context, author log.Author, ref dsref.Ref) error
+}
+
+// assert at compile-time that Logsync is a remote
+var _ remote = (*Logsync)(nil)
+
+func (lsync *Logsync) put(ctx context.Context, author log.Author, r io.Reader) error {
+ if lsync == nil {
+ return ErrNoLogsync
+ }
+
+ if lsync.receiveCheck != nil {
+ // TODO (b5) - need to populate path
+ if err := lsync.receiveCheck(ctx, author, ""); err != nil {
+ return err
+ }
+ }
+
+ data, err := ioutil.ReadAll(r)
+ if err != nil {
+ return err
+ }
+
+ if err := lsync.book.MergeLogBytes(ctx, author, data); err != nil {
+ return err
+ }
+
+ if lsync.didReceive != nil {
+ // TODO (b5) - need to populate path
+ if err := lsync.didReceive(ctx, author, ""); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (lsync *Logsync) get(ctx context.Context, author log.Author, ref dsref.Ref) (log.Author, io.Reader, error) {
+ if lsync == nil {
+ return nil, nil, ErrNoLogsync
+ }
+
+ data, err := lsync.book.LogBytes(ref)
+ if err != nil {
+ return lsync.Author(), nil, err
+ }
+ return lsync.Author(), bytes.NewReader(data), nil
+}
+
+func (lsync *Logsync) del(ctx context.Context, sender log.Author, ref dsref.Ref) error {
+ if lsync == nil {
+ return ErrNoLogsync
+ }
+
+ return lsync.book.RemoveLog(ctx, sender, ref)
+}
+
 // Push is a request to place a log on a remote
 type Push struct {
  ref dsref.Ref

logbook/logsync/logsync_test.go

@@ -108,6 +108,24 @@ func Example() {
  // johnathon has 2 references for basit/nasdaq
 }
 
+func TestNilCallable(t *testing.T) {
+ var logsync *Logsync
+
+ if a := logsync.Author(); a != nil {
+ t.Errorf("author mismatch. expected: '%v', got: '%v' ", nil, a)
+ }
+
+ if _, err := logsync.NewPush(dsref.Ref{}, ""); err != ErrNoLogsync {
+ t.Errorf("error mismatch. expected: '%v', got: '%v' ", ErrNoLogsync, err)
+ }
+ if _, err := logsync.NewPull(dsref.Ref{}, ""); err != ErrNoLogsync {
+ t.Errorf("error mismatch. expected: '%v', got: '%v' ", ErrNoLogsync, err)
+ }
+ if err := logsync.DoRemove(context.Background(), dsref.Ref{}, ""); err != ErrNoLogsync {
+ t.Errorf("error mismatch. expected: '%v', got: '%v' ", ErrNoLogsync, err)
+ }
+}
+
 func makeJohnathonLogbook() *logbook.Book {
  pk, err := decodePk(aPk)
  if err != nil {

remote/client.go

@@ -24,7 +24,7 @@ import (
 )
 
 // ErrNoRemoteClient is returned when no client is allocated
-var ErrNoRemoteClient = fmt.Errorf("not configured to make remote requests")
+var ErrNoRemoteClient = fmt.Errorf("remote: no client to make remote requests")
 
 // Address extracts the address of a remote from a configuration for a given
 // remote name
@@ -73,7 +73,11 @@ func NewClient(node *p2p.QriNode) (*Client, error) {
 
  var ls *logsync.Logsync
  if book := node.Repo.Logbook(); book != nil {
- ls = logsync.New(book)
+ ls = logsync.New(book, func(logsyncConfig *logsync.Options) {
+ if host := node.Host(); host != nil {
+ logsyncConfig.Libp2pHost = host
+ }
+ })
  }
 
  return &Client{
@@ -91,8 +95,8 @@ func (c *Client) CoreAPI() coreiface.CoreAPI {
  return c.capi
 }
 
-// Fetch pulls a logbook from a remote
-func (c *Client) Fetch(ctx context.Context, ref dsref.Ref, remoteAddr string) error {
+// PullLogs pulls logbook data from a remote
+func (c *Client) PullLogs(ctx context.Context, ref dsref.Ref, remoteAddr string) error {
  if c == nil {
  return ErrNoRemoteClient
  }
@@ -109,6 +113,38 @@ func (c *Client) Fetch(ctx context.Context, ref dsref.Ref, remoteAddr string) er
  return pull.Do(ctx)
 }
 
+// PushLogs pushes logbook data to a remote address
+func (c *Client) PushLogs(ctx context.Context, ref dsref.Ref, remoteAddr string) error {
+ if c == nil {
+ return ErrNoRemoteClient
+ }
+
+ if t := addressType(remoteAddr); t == "http" {
+ remoteAddr = remoteAddr + "/remote/logsync"
+ }
+ log.Debugf("pushing logs for %s from %s", ref.Alias(), remoteAddr)
+ push, err := c.logsync.NewPush(ref, remoteAddr)
+ if err != nil {
+ return err
+ }
+
+ return push.Do(ctx)
+}
+
+// RemoveLogs requests a remote remove logbook data from an address
+func (c *Client) RemoveLogs(ctx context.Context, ref dsref.Ref, remoteAddr string) error {
+ if c == nil {
+ return ErrNoRemoteClient
+ }
+
+ if t := addressType(remoteAddr); t == "http" {
+ remoteAddr = remoteAddr + "/remote/logsync"
+ }
+
+ log.Debugf("deleting logs for %s from %s", ref.Alias(), remoteAddr)
+ return c.logsync.DoRemove(ctx, ref, remoteAddr)
+}
+
 // PushDataset pushes the contents of a dataset to a remote
 func (c *Client) PushDataset(ctx context.Context, ref repo.DatasetRef, remoteAddr string) error {
  if c == nil {