Repository: hyperledger/firefly
Pull request #994 (2022-08-20 03:51:24)

Wait for server to exit, before restarting for reset
Fixes https://github.com/hyperledger/firefly/issues/993

We need to wait for the `startFirefly` routine to exit, before we can safely restart it.

This means a second channel to distinguish exit, from an error.

A secondary fix in this PR, is to stop/restart the debug server after the reset - previously it was never stopped.


Patch:
diff --git a/cmd/firefly.go b/cmd/firefly.go
index a4df2b212..2edd25b12 100644
--- a/cmd/firefly.go
+++ b/cmd/firefly.go
@@ -24,6 +24,7 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/gorilla/mux"
 	"github.com/hyperledger/firefly-common/pkg/config"
@@ -118,7 +119,6 @@ func run() error {
 	}
 
 	// Setup signal handling to cancel the context, which shuts down the API Server
-	errChan := make(chan error)
 	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
 
 	for {
@@ -126,7 +126,9 @@ func run() error {
 		rootCtx, rootCancelCtx := context.WithCancel(ctx)
 		mgr := getRootManager()
 		as := apiserver.NewAPIServer()
-		go startFirefly(rootCtx, rootCancelCtx, mgr, as, errChan)
+		errChan := make(chan error, 1)
+		ffDone := make(chan struct{})
+		go startFirefly(rootCtx, rootCancelCtx, mgr, as, errChan, ffDone)
 		select {
 		case sig := <-sigs:
 			log.L(ctx).Infof("Shutting down due to %s", sig.String())
@@ -136,6 +138,8 @@ func run() error {
 		case <-rootCtx.Done():
 			log.L(ctx).Infof("Restarting due to configuration change")
 			mgr.WaitStop()
+			// Must wait for the server to close before we restart
+			<-ffDone
 			// Re-read the configuration
 			resetConfig()
 			if err := config.ReadConfig(configSuffix, cfgFile); err != nil {
@@ -149,9 +153,10 @@ func run() error {
 	}
 }
 
-func startFirefly(ctx context.Context, cancelCtx context.CancelFunc, mgr namespace.Manager, as apiserver.Server, errChan chan error) {
+func startFirefly(ctx context.Context, cancelCtx context.CancelFunc, mgr namespace.Manager, as apiserver.Server, errChan chan error, ffDone chan struct{}) {
 	var err error
 	// Start debug listener
+	var debugServer *http.Server
 	debugPort := config.GetInt(coreconfig.DebugPort)
 	if debugPort >= 0 {
 		r := mux.NewRouter()
@@ -160,12 +165,20 @@ func startFirefly(ctx context.Context, cancelCtx context.CancelFunc, mgr namespa
 		r.PathPrefix("/debug/pprof/symbol").HandlerFunc(pprof.Symbol)
 		r.PathPrefix("/debug/pprof/trace").HandlerFunc(pprof.Trace)
 		r.PathPrefix("/debug/pprof/").HandlerFunc(pprof.Index)
+		debugServer = &http.Server{Addr: fmt.Sprintf("localhost:%d", debugPort), Handler: r, ReadHeaderTimeout: 30 * time.Second}
 		go func() {
-			_ = http.ListenAndServe(fmt.Sprintf("localhost:%d", debugPort), r)
+			_ = debugServer.ListenAndServe()
 		}()
 		log.L(ctx).Debugf("Debug HTTP endpoint listening on localhost:%d", debugPort)
 	}
 
+	defer func() {
+		if debugServer != nil {
+			_ = debugServer.Close()
+		}
+		close(ffDone)
+	}()
+
 	if err = mgr.Init(ctx, cancelCtx); err != nil {
 		errChan <- err
 		return
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index a00e384bb..da5f799f6 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -19578,7 +19578,7 @@ paths:
       - Non-Default Namespace
   /namespaces/{ns}/network/organizations/{nameOrId}:
     get:
-      description: Gets information about a specific org in the network
+      description: Gets information about a specifc org in the network
       operationId: getNetworkOrgNamespace
       parameters:
       - description: The name or ID of the org
@@ -20409,64 +20409,6 @@ paths:
           description: ""
       tags:
       - Non-Default Namespace
-  /namespaces/{ns}/pins/rewind:
-    post:
-      description: Force a rewind of the event aggregator to a previous position,
-        to re-evaluate all unconfirmed pins since that point
-      operationId: postPinsRewindNamespace
-      parameters:
-      - description: The namespace which scopes this request
-        in: path
-        name: ns
-        required: true
-        schema:
-          example: default
-          type: string
-      - description: Server-side request timeout (milliseconds, or set a custom suffix
-          like 10s)
-        in: header
-        name: Request-Timeout
-        schema:
-          default: 2m0s
-          type: string
-      requestBody:
-        content:
-          application/json:
-            schema:
-              properties:
-                batch:
-                  description: The ID of the batch to which the event aggregator should
-                    rewind. Either sequence or batch must be specified
-                  format: uuid
-                  type: string
-                sequence:
-                  description: The sequence of the pin to which the event aggregator
-                    should rewind. Either sequence or batch must be specified
-                  format: int64
-                  type: integer
-              type: object
-      responses:
-        "200":
-          content:
-            application/json:
-              schema:
-                properties:
-                  batch:
-                    description: The ID of the batch to which the event aggregator
-                      should rewind. Either sequence or batch must be specified
-                    format: uuid
-                    type: string
-                  sequence:
-                    description: The sequence of the pin to which the event aggregator
-                      should rewind. Either sequence or batch must be specified
-                    format: int64
-                    type: integer
-                type: object
-          description: Success
-        default:
-          description: ""
-      tags:
-      - Non-Default Namespace
   /namespaces/{ns}/status:
     get:
       description: Gets the status of this namespace
@@ -27246,7 +27188,7 @@ paths:
       - Default Namespace
   /network/organizations/{nameOrId}:
     get:
-      description: Gets information about a specific org in the network
+      description: Gets information about a specifc org in the network
       operationId: getNetworkOrg
       parameters:
       - description: The name or ID of the org
@@ -28035,57 +27977,6 @@ paths:
           description: ""
       tags:
       - Default Namespace
-  /pins/rewind:
-    post:
-      description: Force a rewind of the event aggregator to a previous position,
-        to re-evaluate all unconfirmed pins since that point
-      operationId: postPinsRewind
-      parameters:
-      - description: Server-side request timeout (milliseconds, or set a custom suffix
-          like 10s)
-        in: header
-        name: Request-Timeout
-        schema:
-          default: 2m0s
-          type: string
-      requestBody:
-        content:
-          application/json:
-            schema:
-              properties:
-                batch:
-                  description: The ID of the batch to which the event aggregator should
-                    rewind. Either sequence or batch must be specified
-                  format: uuid
-                  type: string
-                sequence:
-                  description: The sequence of the pin to which the event aggregator
-                    should rewind. Either sequence or batch must be specified
-                  format: int64
-                  type: integer
-              type: object
-      responses:
-        "200":
-          content:
-            application/json:
-              schema:
-                properties:
-                  batch:
-                    description: The ID of the batch to which the event aggregator
-                      should rewind. Either sequence or batch must be specified
-                    format: uuid
-                    type: string
-                  sequence:
-                    description: The sequence of the pin to which the event aggregator
-                      should rewind. Either sequence or batch must be specified
-                    format: int64
-                    type: integer
-                type: object
-          description: Success
-        default:
-          description: ""
-      tags:
-      - Default Namespace
   /status:
     get:
       description: Gets the status of this namespace
diff --git a/internal/apiserver/route_post_pins_rewind.go b/internal/apiserver/route_post_pins_rewind.go
deleted file mode 100644
index 57e3cb130..000000000
--- a/internal/apiserver/route_post_pins_rewind.go
+++ /dev/null
@@ -1,42 +0,0 @@
-// Copyright © 2022 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// 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 apiserver
-
-import (
-	"net/http"
-
-	"github.com/hyperledger/firefly-common/pkg/ffapi"
-	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/pkg/core"
-)
-
-var postPinsRewind = &ffapi.Route{
-	Name:            "postPinsRewind",
-	Path:            "pins/rewind",
-	Method:          http.MethodPost,
-	PathParams:      nil,
-	QueryParams:     nil,
-	Description:     coremsgs.APIEndpointsPostPinsRewind,
-	JSONInputValue:  func() interface{} { return &core.PinRewind{} },
-	JSONOutputValue: func() interface{} { return &core.PinRewind{} },
-	JSONOutputCodes: []int{http.StatusOK},
-	Extensions: &coreExtensions{
-		CoreJSONHandler: func(r *ffapi.APIRequest, cr *coreRequest) (output interface{}, err error) {
-			return cr.or.RewindPins(cr.ctx, r.Input.(*core.PinRewind))
-		},
-	},
-}
diff --git a/internal/apiserver/routes.go b/internal/apiserver/routes.go
index 9550a71ec..537907036 100644
--- a/internal/apiserver/routes.go
+++ b/internal/apiserver/routes.go
@@ -143,7 +143,6 @@ var routes = append(
 		postNewOrganizationSelf,
 		postNodesSelf,
 		postOpRetry,
-		postPinsRewind,
 		postTokenApproval,
 		postTokenBurn,
 		postTokenMint,
diff --git a/internal/coremsgs/en_api_translations.go b/internal/coremsgs/en_api_translations.go
index 0895c3a23..ae702515b 100644
--- a/internal/coremsgs/en_api_translations.go
+++ b/internal/coremsgs/en_api_translations.go
@@ -113,7 +113,7 @@ var (
 	APIEndpointsGetNetworkIdentities            = ffm("api.endpoints.getNetworkIdentities", "Gets the list of identities in the network (deprecated - use /identities instead of /network/identities")
 	APIEndpointsGetNetworkNode                  = ffm("api.endpoints.getNetworkNode", "Gets information about a specific node in the network")
 	APIEndpointsGetNetworkNodes                 = ffm("api.endpoints.getNetworkNodes", "Gets a list of nodes in the network")
-	APIEndpointsGetNetworkOrg                   = ffm("api.endpoints.getNetworkOrg", "Gets information about a specific org in the network")
+	APIEndpointsGetNetworkOrg                   = ffm("api.endpoints.getNetworkOrg", "Gets information about a specifc org in the network")
 	APIEndpointsGetNetworkOrgs                  = ffm("api.endpoints.APIEndpointsGetNetworkOrgs", "Gets a list of orgs in the network")
 	APIEndpointsGetOpByID                       = ffm("api.endpoints.getOpByID", "Gets an operation by ID")
 	APIEndpointsGetOps                          = ffm("api.endpoints.getOps", "Gets a a list of operations")
@@ -162,7 +162,6 @@ var (
 	APIEndpointsPostNewOrganization             = ffm("api.endpoints.postNewOrganization", "Registers a new org in the network")
 	APIEndpointsPostNewSubscription             = ffm("api.endpoints.postNewSubscription", "Creates a new subscription for an application to receive events from FireFly")
 	APIEndpointsPostOpRetry                     = ffm("api.endpoints.postOpRetry", "Retries a failed operation")
-	APIEndpointsPostPinsRewind                  = ffm("api.endpoints.postPinsRewind", "Force a rewind of the event aggregator to a previous position, to re-evaluate all unconfirmed pins since that point")
 	APIEndpointsPostTokenApproval               = ffm("api.endpoints.postTokenApproval", "Creates a token approval")
 	APIEndpointsPostTokenBurn                   = ffm("api.endpoints.postTokenBurn", "Burns some tokens")
 	APIEndpointsPostTokenMint                   = ffm("api.endpoints.postTokenMint", "Mints some tokens")
diff --git a/internal/coremsgs/en_struct_descriptions.go b/internal/coremsgs/en_struct_descriptions.go
index 2f988d58c..1921479a5 100644
--- a/internal/coremsgs/en_struct_descriptions.go
+++ b/internal/coremsgs/en_struct_descriptions.go
@@ -452,18 +452,16 @@ var (
 	BatchFlushStatusTotalErrors          = ffm("BatchFlushStatus.totalErrors", "The total count of error flushed encountered by this processor since it started")
 
 	// Pin field descriptions
-	PinSequence       = ffm("Pin.sequence", "The order of the pin in the local FireFly database, which matches the order in which pins were delivered to FireFly by the blockchain connector event stream")
-	PinNamespace      = ffm("Pin.namespace", "The namespace of the pin")
-	PinMasked         = ffm("Pin.masked", "True if the pin is for a private message, and hence is masked with the group ID and salted with a nonce so observers of the blockchain cannot use pin hash to match this transaction to other transactions or participants")
-	PinHash           = ffm("Pin.hash", "The hash represents a topic within a message in the batch. If a message has multiple topics, then multiple pins are created. If the message is private, the hash is masked for privacy")
-	PinBatch          = ffm("Pin.batch", "The UUID of the batch of messages this pin is part of")
-	PinBatchHash      = ffm("Pin.batchHash", "The manifest hash batch of messages this pin is part of")
-	PinIndex          = ffm("Pin.index", "The index of this pin within the batch. One pin is created for each topic, of each message in the batch")
-	PinDispatched     = ffm("Pin.dispatched", "Once true, this pin has been processed and will not be processed again")
-	PinSigner         = ffm("Pin.signer", "The blockchain signing key that submitted this transaction, as passed through to FireFly by the smart contract that emitted the blockchain event")
-	PinCreated        = ffm("Pin.created", "The time the FireFly node created the pin")
-	PinRewindSequence = ffm("PinRewind.sequence", "The sequence of the pin to which the event aggregator should rewind. Either sequence or batch must be specified")
-	PinRewindBatch    = ffm("PinRewind.batch", "The ID of the batch to which the event aggregator should rewind. Either sequence or batch must be specified")
+	PinSequence   = ffm("Pin.sequence", "The order of the pin in the local FireFly database, which matches the order in which pins were delivered to FireFly by the blockchain connector event stream")
+	PinNamespace  = ffm("Pin.namespace", "The namespace of the pin")
+	PinMasked     = ffm("Pin.masked", "True if the pin is for a private message, and hence is masked with the group ID and salted with a nonce so observers of the blockchain cannot use pin hash to match this transaction to other transactions or participants")
+	PinHash       = ffm("Pin.hash", "The hash represents a topic within a message in the batch. If a message has multiple topics, then multiple pins are created. If the message is private, the hash is masked for privacy")
+	PinBatch      = ffm("Pin.batch", "The UUID of the batch of messages this pin is part of")
+	PinBatchHash  = ffm("Pin.batchHash", "The manifest hash batch of messages this pin is part of")
+	PinIndex      = ffm("Pin.index", "The index of this pin within the batch. One pin is created for each topic, of each message in the batch")
+	PinDispatched = ffm("Pin.dispatched", "Once true, this pin has been processed and will not be processed again")
+	PinSigner     = ffm("Pin.signer", "The blockchain signing key that submitted this transaction, as passed through to FireFly by the smart contract that emitted the blockchain event")
+	PinCreated    = ffm("Pin.created", "The time the FireFly node created the pin")
 
 	// Subscription field descriptions
 	SubscriptionID        = ffm("Subscription.id", "The UUID of the subscription")
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index 8ade6c51e..6d3b84fa6 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -62,7 +62,6 @@ type EventManager interface {
 	DeleteDurableSubscription(ctx context.Context, subDef *core.Subscription) (err error)
 	CreateUpdateDurableSubscription(ctx context.Context, subDef *core.Subscription, mustNew bool) (err error)
 	EnrichEvent(ctx context.Context, event *core.Event) (*core.EnrichedEvent, error)
-	QueueBatchRewind(batchID *fftypes.UUID)
 	Start() error
 	WaitStop()
 
@@ -280,7 +279,3 @@ func (em *eventManager) GetPlugins() []*core.NamespaceStatusPlugin {
 func (em *eventManager) EnrichEvent(ctx context.Context, event *core.Event) (*core.EnrichedEvent, error) {
 	return em.enricher.enrichEvent(ctx, event)
 }
-
-func (em *eventManager) QueueBatchRewind(batchID *fftypes.UUID) {
-	em.aggregator.queueBatchRewind(batchID)
-}
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 0e4f64d73..c6c1c42ff 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -108,7 +108,6 @@ type Orchestrator interface {
 	GetBlockchainEventByID(ctx context.Context, id string) (*core.BlockchainEvent, error)
 	GetBlockchainEvents(ctx context.Context, filter database.AndFilter) ([]*core.BlockchainEvent, *database.FilterResult, error)
 	GetPins(ctx context.Context, filter database.AndFilter) ([]*core.Pin, *database.FilterResult, error)
-	RewindPins(ctx context.Context, rewind *core.PinRewind) (*core.PinRewind, error)
 
 	// Charts
 	GetChartHistogram(ctx context.Context, startTime int64, endTime int64, buckets int64, tableName database.CollectionName) ([]*core.ChartHistogram, error)
@@ -525,19 +524,3 @@ func (or *orchestrator) Authorize(ctx context.Context, authReq *fftypes.AuthReq)
 	}
 	return nil
 }
-
-func (or *orchestrator) RewindPins(ctx context.Context, rewind *core.PinRewind) (*core.PinRewind, error) {
-	if rewind.Sequence > 0 {
-		fb := database.PinQueryFactory.NewFilter(ctx)
-		if pins, _, err := or.GetPins(ctx, fb.And(fb.Eq("seq", rewind.Sequence))); err != nil {
-			return nil, err
-		} else if len(pins) > 0 {
-			rewind.Batch = pins[0].Batch
-			or.events.QueueBatchRewind(rewind.Batch)
-			return rewind, nil
-		}
-		return nil, i18n.NewError(ctx, coremsgs.Msg404NotFound)
-	}
-	or.events.QueueBatchRewind(rewind.Batch)
-	return rewind, nil
-}
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index a2fa147df..723f750ab 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -341,13 +341,13 @@ func (_m *Plugin) Start() error {
 	return r0
 }
 
-// SubmitBatchPin provides a mock function with given fields: ctx, nsOpID, networkNamespace, signingKey, batch, location
-func (_m *Plugin) SubmitBatchPin(ctx context.Context, nsOpID string, networkNamespace string, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
-	ret := _m.Called(ctx, nsOpID, networkNamespace, signingKey, batch, location)
+// SubmitBatchPin provides a mock function with given fields: ctx, nsOpID, remtoeNamespace, signingKey, batch, location
+func (_m *Plugin) SubmitBatchPin(ctx context.Context, nsOpID string, remtoeNamespace string, signingKey string, batch *blockchain.BatchPin, location *fftypes.JSONAny) error {
+	ret := _m.Called(ctx, nsOpID, remtoeNamespace, signingKey, batch, location)
 
 	var r0 error
 	if rf, ok := ret.Get(0).(func(context.Context, string, string, string, *blockchain.BatchPin, *fftypes.JSONAny) error); ok {
-		r0 = rf(ctx, nsOpID, networkNamespace, signingKey, batch, location)
+		r0 = rf(ctx, nsOpID, remtoeNamespace, signingKey, batch, location)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index 4dc55fb28..2689ff43d 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -219,11 +219,6 @@ func (_m *EventManager) NewSubscriptions() chan<- *fftypes.UUID {
 	return r0
 }
 
-// QueueBatchRewind provides a mock function with given fields: batchID
-func (_m *EventManager) QueueBatchRewind(batchID *fftypes.UUID) {
-	_m.Called(batchID)
-}
-
 // SharedStorageBatchDownloaded provides a mock function with given fields: ss, payloadRef, data
 func (_m *EventManager) SharedStorageBatchDownloaded(ss sharedstorage.Plugin, payloadRef string, data []byte) (*fftypes.UUID, error) {
 	ret := _m.Called(ss, payloadRef, data)
diff --git a/mocks/orchestratormocks/orchestrator.go b/mocks/orchestratormocks/orchestrator.go
index 35e9c8cca..9cfcf99ae 100644
--- a/mocks/orchestratormocks/orchestrator.go
+++ b/mocks/orchestratormocks/orchestrator.go
@@ -1223,29 +1223,6 @@ func (_m *Orchestrator) RequestReply(ctx context.Context, msg *core.MessageInOut
 	return r0, r1
 }
 
-// RewindPins provides a mock function with given fields: ctx, rewind
-func (_m *Orchestrator) RewindPins(ctx context.Context, rewind *core.PinRewind) (*core.PinRewind, error) {
-	ret := _m.Called(ctx, rewind)
-
-	var r0 *core.PinRewind
-	if rf, ok := ret.Get(0).(func(context.Context, *core.PinRewind) *core.PinRewind); ok {
-		r0 = rf(ctx, rewind)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*core.PinRewind)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *core.PinRewind) error); ok {
-		r1 = rf(ctx, rewind)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // Start provides a mock function with given fields:
 func (_m *Orchestrator) Start() error {
 	ret := _m.Called()
diff --git a/pkg/core/pin.go b/pkg/core/pin.go
index b5bcc2c02..62c8d23c0 100644
--- a/pkg/core/pin.go
+++ b/pkg/core/pin.go
@@ -54,8 +54,3 @@ type Pin struct {
 func (p *Pin) LocalSequence() int64 {
 	return p.Sequence
 }
-
-type PinRewind struct {
-	Sequence int64         `ffstruct:"PinRewind" json:"sequence"`
-	Batch    *fftypes.UUID `ffstruct:"PinRewind" json:"batch"`
-}


Test patch:
diff --git a/cmd/firefly_test.go b/cmd/firefly_test.go
index 167fa389f..363005621 100644
--- a/cmd/firefly_test.go
+++ b/cmd/firefly_test.go
@@ -102,8 +102,9 @@ func TestExecOkRestartThenExit(t *testing.T) {
 		initCount++
 		if initCount == 2 {
 			init.ReturnArguments = mock.Arguments{fmt.Errorf("second run")}
+		} else {
+			cancelOrContext()
 		}
-		cancelOrContext()
 	}
 	o.On("Start").Return(nil)
 	ws := o.On("WaitStop")
@@ -151,7 +152,7 @@ func TestAPIServerError(t *testing.T) {
 	as.On("Serve", mock.Anything, o).Return(fmt.Errorf("pop"))
 
 	errChan := make(chan error)
-	go startFirefly(context.Background(), func() {}, o, as, errChan)
+	go startFirefly(context.Background(), func() {}, o, as, errChan, make(chan struct{}))
 	err := <-errChan
 	assert.EqualError(t, err, "pop")
 }
diff --git a/internal/apiserver/route_post_pins_rewind_test.go b/internal/apiserver/route_post_pins_rewind_test.go
deleted file mode 100644
index 2fbbc3379..000000000
--- a/internal/apiserver/route_post_pins_rewind_test.go
+++ /dev/null
@@ -1,45 +0,0 @@
-// Copyright © 2021 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// 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 apiserver
-
-import (
-	"bytes"
-	"encoding/json"
-	"net/http/httptest"
-	"testing"
-
-	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
-)
-
-func TestPostPinsRewind(t *testing.T) {
-	o, r := newTestAPIServer()
-	o.On("Authorize", mock.Anything, mock.Anything).Return(nil)
-	input := core.PinRewind{}
-	var buf bytes.Buffer
-	json.NewEncoder(&buf).Encode(&input)
-	req := httptest.NewRequest("POST", "/api/v1/pins/rewind", &buf)
-	req.Header.Set("Content-Type", "application/json; charset=utf-8")
-	res := httptest.NewRecorder()
-
-	o.On("RewindPins", mock.Anything, mock.AnythingOfType("*core.PinRewind")).
-		Return(&core.PinRewind{}, nil)
-	r.ServeHTTP(res, req)
-
-	assert.Equal(t, 200, res.Result().StatusCode)
-}
diff --git a/internal/events/aggregator_rewind_test.go b/internal/events/aggregator_rewind_test.go
index 23c5e5943..4a53c871f 100644
--- a/internal/events/aggregator_rewind_test.go
+++ b/internal/events/aggregator_rewind_test.go
@@ -217,15 +217,15 @@ func TestProcessStagedRewindsNoDIDs(t *testing.T) {
 
 func TestPopRewindsDoublePopNoBlock(t *testing.T) {
 
-	em, done := newTestEventManager(t)
-	defer done()
+	ag, cancel := newTestAggregator()
+	defer cancel()
 
-	em.QueueBatchRewind(fftypes.NewUUID())
-	batchIDs := em.aggregator.rewinder.popRewinds()
+	ag.rewinder.queuedRewinds = []*rewind{{rewindType: rewindBatch}}
+	batchIDs := ag.rewinder.popRewinds()
 	assert.Empty(t, batchIDs)
 
-	em.QueueBatchRewind(fftypes.NewUUID())
-	batchIDs = em.aggregator.rewinder.popRewinds()
+	ag.rewinder.queuedRewinds = []*rewind{{rewindType: rewindBatch}}
+	batchIDs = ag.rewinder.popRewinds()
 	assert.Empty(t, batchIDs)
 
 }
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index 61ed0815b..f4d6c11f7 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -477,54 +477,3 @@ func TestAuthorizeNoPlugin(t *testing.T) {
 	err := or.Authorize(context.Background(), &fftypes.AuthReq{})
 	assert.NoError(t, err)
 }
-
-func TestRewindPinsSeq(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	rewind := &core.PinRewind{Sequence: 100}
-	batchID := fftypes.NewUUID()
-
-	or.mdi.On("GetPins", mock.Anything, "ns", mock.Anything).Return([]*core.Pin{{Batch: batchID}}, nil, nil)
-	or.mem.On("QueueBatchRewind", batchID).Return()
-
-	result, err := or.RewindPins(context.Background(), rewind)
-	assert.NoError(t, err)
-	assert.Equal(t, int64(100), result.Sequence)
-	assert.Equal(t, batchID, result.Batch)
-}
-
-func TestRewindPinsFail(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	rewind := &core.PinRewind{Sequence: 100}
-
-	or.mdi.On("GetPins", mock.Anything, "ns", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
-
-	_, err := or.RewindPins(context.Background(), rewind)
-	assert.EqualError(t, err, "pop")
-}
-
-func TestRewindPinsNotFound(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	rewind := &core.PinRewind{Sequence: 100}
-
-	or.mdi.On("GetPins", mock.Anything, "ns", mock.Anything).Return(nil, nil, nil)
-
-	_, err := or.RewindPins(context.Background(), rewind)
-	assert.Regexp(t, "FF10109", err)
-}
-
-func TestRewindPinsBatch(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	batchID := fftypes.NewUUID()
-	rewind := &core.PinRewind{Batch: batchID}
-
-	or.mem.On("QueueBatchRewind", batchID).Return()
-
-	result, err := or.RewindPins(context.Background(), rewind)
-	assert.NoError(t, err)
-	assert.Equal(t, int64(0), result.Sequence)
-	assert.Equal(t, batchID, result.Batch)
-}