Finalize segfetcher module

go/lib/infra/modules/segfetcher/BUILD.bazel

@@ -3,12 +3,15 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "doc.go",
+        "fetcher.go",
         "request.go",
         "requester.go",
         "resolver.go",
         "segreplyhandler.go",
         "segs.go",
         "splitter.go",
+        "validator.go",
     ],
     importpath = "github.com/scionproto/scion/go/lib/infra/modules/segfetcher",
     visibility = ["//visibility:public"],
@@ -24,16 +27,17 @@ go_library(
         "//go/lib/pathdb:go_default_library",
         "//go/lib/pathdb/query:go_default_library",
         "//go/lib/revcache:go_default_library",
+        "//go/proto:go_default_library",
     ],
 )
 
 go_test(
     name = "go_default_test",
     srcs = [
+        "fetcher_test.go",
         "requester_test.go",
         "resolver_test.go",
         "segreplyhandler_test.go",
-        "splitter_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
@@ -42,15 +46,17 @@ go_test(
         "//go/lib/ctrl/path_mgmt:go_default_library",
         "//go/lib/ctrl/seg:go_default_library",
         "//go/lib/infra:go_default_library",
-        "//go/lib/infra/mock_infra:go_default_library",
         "//go/lib/infra/modules/segfetcher/mock_segfetcher:go_default_library",
         "//go/lib/infra/modules/segverifier:go_default_library",
+        "//go/lib/mocks/net/mock_net:go_default_library",
+        "//go/lib/pathdb/mock_pathdb:go_default_library",
         "//go/lib/pathdb/query:go_default_library",
         "//go/lib/xtest:go_default_library",
         "//go/lib/xtest/graph:go_default_library",
         "//go/lib/xtest/matchers:go_default_library",
         "//go/proto:go_default_library",
         "@com_github_golang_mock//gomock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
     ],
 )

go/lib/infra/modules/segfetcher/doc.go

@@ -0,0 +1,18 @@
+// Copyright 2019 Anapaya Systems
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package segfetcher contains all the logic that is needed to fetch segments,
+// verify and store segments in an efficient manner. It is designed to be
+// pluggable into sciond and PS.
+package segfetcher

go/lib/infra/modules/segfetcher/fetcher.go

@@ -0,0 +1,159 @@
+// Copyright 2019 Anapaya Systems
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package segfetcher
+
+import (
+	"context"
+	"net"
+	"time"
+
+	"github.com/scionproto/scion/go/lib/addr"
+	"github.com/scionproto/scion/go/lib/common"
+	"github.com/scionproto/scion/go/lib/infra"
+	"github.com/scionproto/scion/go/lib/log"
+	"github.com/scionproto/scion/go/lib/pathdb"
+	"github.com/scionproto/scion/go/lib/revcache"
+)
+
+// FetcherConfig is the configuration for the fetcher.
+type FetcherConfig struct {
+	// QueryInterval specifies after how much time segments should be
+	// refetched at the remote server.
+	QueryInterval time.Duration
+	// LocalIA is the IA this process is in.
+	LocalIA addr.IA
+	// VerificationFactory is the verification factory to use.
+	VerificationFactory infra.VerificationFactory
+	// ASInspector is the as inspector to use.
+	ASInspector infra.ASInspector
+	// PathDB is the path db to use.
+	PathDB pathdb.PathDB
+	// RevCache is the revocation cache to use.
+	RevCache revcache.RevCache
+	// Messenger is the messenger to use.
+	Messenger infra.Messenger
+	// DstProvider provides destinations to fetch segments from
+	DstProvider DstProvider
+	// Validator is used to validate requests.
+	Validator Validator
+	// Splitter is used to split requests.
+	Splitter Splitter
+	// CryptoLookupAtLocalCS indicates whether crypto to verify path material
+	// should be fetched from the local CS or from the sender of the path
+	// material.
+	CryptoLookupAtLocalCS bool
+}
+
+// New creates a new fetcher from the configuration.
+func (cfg FetcherConfig) New() *Fetcher {
+	return &Fetcher{
+		Validator: cfg.Validator,
+		Splitter:  cfg.Splitter,
+		Resolver:  NewResolver(cfg.PathDB),
+		Requester: &DefaultRequester{API: cfg.Messenger, DstProvider: cfg.DstProvider},
+		ReplyHandler: &SegReplyHandler{
+			Verifier: &SegVerifier{Verifier: cfg.VerificationFactory.NewVerifier()},
+			Storage:  &DefaultStorage{PathDB: cfg.PathDB, RevCache: cfg.RevCache},
+		},
+		PathDB:                cfg.PathDB,
+		QueryInterval:         cfg.QueryInterval,
+		CryptoLookupAtLocalCS: cfg.CryptoLookupAtLocalCS,
+	}
+}
+
+// Fetcher fetches, verifies and stores segments for a given path request.
+type Fetcher struct {
+	Validator             Validator
+	Splitter              Splitter
+	Resolver              Resolver
+	Requester             Requester
+	ReplyHandler          ReplyHandler
+	PathDB                pathdb.PathDB
+	QueryInterval         time.Duration
+	CryptoLookupAtLocalCS bool
+}
+
+// FetchSegs fetches the required segments to build a path between src and dst
+// of the request. First the request is validated and then depending on the
+// cache the segments are fetched from the remote server.
+func (f *Fetcher) FetchSegs(ctx context.Context, req Request) (Segments, error) {
+	if f.Validator != nil {
+		if err := f.Validator.Validate(req); err != nil {
+			return Segments{}, err
+		}
+	}
+	reqSet, err := f.Splitter.Split(ctx, req)
+	if err != nil {
+		return Segments{}, err
+	}
+	var segs Segments
+	i := 0
+	for {
+		log.FromCtx(ctx).Trace("Request to process", "req", reqSet)
+		segs, reqSet, err = f.Resolver.Resolve(ctx, segs, reqSet)
+		if err != nil {
+			return Segments{}, err
+		}
+		log.FromCtx(ctx).Trace("After resolving", "req", reqSet)
+		if reqSet.IsEmpty() {
+			break
+		}
+		if i > 3 {
+			log.FromCtx(ctx).Crit("No convergence in looking up", "i", i)
+			return segs, common.NewBasicError("Segment lookup doesn't converge", nil,
+				"iterations", i)
+		}
+		replies := f.Requester.Request(ctx, reqSet)
+		// TODO(lukedirtwalker): We need to have early trigger for the last request.
+		if err := f.waitOnProcessed(ctx, replies); err != nil {
+			return Segments{}, err
+		}
+	}
+	return segs, nil
+}
+
+func (f *Fetcher) waitOnProcessed(ctx context.Context, replies <-chan ReplyOrErr) error {
+	for reply := range replies {
+		// TODO(lukedirtwalker): Should we do this in go routines?
+		if reply.Err != nil {
+			return reply.Err
+		}
+		if reply.Reply == nil || reply.Reply.Recs == nil {
+			continue
+		}
+		r := f.ReplyHandler.Handle(ctx, reply.Reply, f.verifyServer(reply), nil)
+		select {
+		case <-r.FullReplyProcessed():
+			if err := r.Err(); err != nil {
+				return err
+			}
+			_, err := f.PathDB.InsertNextQuery(ctx, reply.Req.Src, reply.Req.Dst, nil,
+				time.Now().Add(f.QueryInterval))
+			if err != nil {
+				log.FromCtx(ctx).Warn("Failed to insert next query", "err", err)
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return nil
+}
+
+func (f *Fetcher) verifyServer(reply ReplyOrErr) net.Addr {
+	if f.CryptoLookupAtLocalCS {
+		return nil
+	}
+	return reply.Peer
+}

go/lib/infra/modules/segfetcher/fetcher_test.go

@@ -0,0 +1,133 @@
+// Copyright 2019 Anapaya Systems
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package segfetcher_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/scionproto/scion/go/lib/ctrl/seg"
+	"github.com/scionproto/scion/go/lib/infra/modules/segfetcher"
+	"github.com/scionproto/scion/go/lib/infra/modules/segfetcher/mock_segfetcher"
+	"github.com/scionproto/scion/go/lib/pathdb/mock_pathdb"
+)
+
+type TestableFetcher struct {
+	Validator     *mock_segfetcher.MockValidator
+	Splitter      *mock_segfetcher.MockSplitter
+	Resolver      *mock_segfetcher.MockResolver
+	Requester     *mock_segfetcher.MockRequester
+	ReplyHandler  *mock_segfetcher.MockReplyHandler
+	PathDB        *mock_pathdb.MockPathDB
+	QueryInterval time.Duration
+}
+
+func NewTestFetcher(ctrl *gomock.Controller) *TestableFetcher {
+	return &TestableFetcher{
+		Validator:     mock_segfetcher.NewMockValidator(ctrl),
+		Splitter:      mock_segfetcher.NewMockSplitter(ctrl),
+		Resolver:      mock_segfetcher.NewMockResolver(ctrl),
+		Requester:     mock_segfetcher.NewMockRequester(ctrl),
+		ReplyHandler:  mock_segfetcher.NewMockReplyHandler(ctrl),
+		PathDB:        mock_pathdb.NewMockPathDB(ctrl),
+		QueryInterval: time.Minute,
+	}
+}
+
+func (f *TestableFetcher) Fetcher() *segfetcher.Fetcher {
+	return &segfetcher.Fetcher{
+		Validator:     f.Validator,
+		Splitter:      f.Splitter,
+		Resolver:      f.Resolver,
+		Requester:     f.Requester,
+		ReplyHandler:  f.ReplyHandler,
+		PathDB:        f.PathDB,
+		QueryInterval: f.QueryInterval,
+	}
+}
+
+func TestFetcher(t *testing.T) {
+	rootCtrl := gomock.NewController(t)
+	defer rootCtrl.Finish()
+	tg := newTestGraph(rootCtrl)
+	testErr := errors.New("Test err")
+
+	tests := map[string]struct {
+		PrepareFetcher func(*TestableFetcher)
+		Request        segfetcher.Request
+		ErrorAssertion require.ErrorAssertionFunc
+		ExpectedSegs   segfetcher.Segments
+	}{
+		"Invalid request": {
+			PrepareFetcher: func(f *TestableFetcher) {
+				f.Validator.EXPECT().Validate(gomock.Any()).Return(testErr)
+			},
+			ErrorAssertion: require.Error,
+		},
+		"Splitter error": {
+			PrepareFetcher: func(f *TestableFetcher) {
+				f.Validator.EXPECT().Validate(gomock.Any())
+				f.Splitter.EXPECT().Split(gomock.Any(), gomock.Any()).
+					Return(segfetcher.RequestSet{}, testErr)
+			},
+			ErrorAssertion: require.Error,
+		},
+		"Resolver error": {
+			PrepareFetcher: func(f *TestableFetcher) {
+				f.Validator.EXPECT().Validate(gomock.Any())
+				f.Splitter.EXPECT().Split(gomock.Any(), gomock.Any())
+				f.Resolver.EXPECT().Resolve(gomock.Any(), gomock.Any(), gomock.Any()).
+					Return(segfetcher.Segments{}, segfetcher.RequestSet{}, testErr)
+			},
+			ErrorAssertion: require.Error,
+		},
+		"Immediately resolved": {
+			PrepareFetcher: func(f *TestableFetcher) {
+				f.Validator.EXPECT().Validate(gomock.Any())
+				reqSet := segfetcher.RequestSet{
+					Up: segfetcher.Request{Src: non_core_111, Dst: core_130},
+				}
+				f.Splitter.EXPECT().Split(gomock.Any(), gomock.Any()).
+					Return(reqSet, nil)
+				f.Resolver.EXPECT().Resolve(gomock.Any(), gomock.Any(), gomock.Eq(reqSet)).
+					Return(segfetcher.Segments{Up: seg.Segments{tg.seg130_111}},
+						segfetcher.RequestSet{}, nil)
+			},
+			ErrorAssertion: require.NoError,
+			ExpectedSegs:   segfetcher.Segments{Up: seg.Segments{tg.seg130_111}},
+		},
+		// XXX(lukedirtwalker): testing the full loop is quite involved, and is
+		// therefore currently omitted.
+	}
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+			ctx, cancelF := context.WithTimeout(context.Background(), time.Second)
+			defer cancelF()
+			f := NewTestFetcher(ctrl)
+			test.PrepareFetcher(f)
+			segs, err := f.Fetcher().FetchSegs(ctx, test.Request)
+			test.ErrorAssertion(t, err)
+			assert.Equal(t, test.ExpectedSegs, segs)
+		})
+	}
+}

go/lib/infra/modules/segfetcher/mock_segfetcher/BUILD.bazel

@@ -6,11 +6,9 @@ go_library(
     importpath = "github.com/scionproto/scion/go/lib/infra/modules/segfetcher/mock_segfetcher",
     visibility = ["//visibility:public"],
     deps = [
-        "//go/lib/addr:go_default_library",
         "//go/lib/ctrl/path_mgmt:go_default_library",
         "//go/lib/infra/modules/segfetcher:go_default_library",
         "//go/lib/infra/modules/segverifier:go_default_library",
-        "//go/lib/pathdb/query:go_default_library",
         "@com_github_golang_mock//gomock:go_default_library",
     ],
 )