Repository: hyperledger/firefly
Pull request #991 (2022-08-19 19:27:31)

Add /pins/rewind API for requesing manual rewind
Mitigation for https://github.com/hyperledger/firefly/issues/986 and similar issues.


Patch:
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index da5f799f6..a00e384bb 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 specifc org in the network
+      description: Gets information about a specific org in the network
       operationId: getNetworkOrgNamespace
       parameters:
       - description: The name or ID of the org
@@ -20409,6 +20409,64 @@ 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
@@ -27188,7 +27246,7 @@ paths:
       - Default Namespace
   /network/organizations/{nameOrId}:
     get:
-      description: Gets information about a specifc org in the network
+      description: Gets information about a specific org in the network
       operationId: getNetworkOrg
       parameters:
       - description: The name or ID of the org
@@ -27977,6 +28035,57 @@ 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
new file mode 100644
index 000000000..57e3cb130
--- /dev/null
+++ b/internal/apiserver/route_post_pins_rewind.go
@@ -0,0 +1,42 @@
+// 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 537907036..9550a71ec 100644
--- a/internal/apiserver/routes.go
+++ b/internal/apiserver/routes.go
@@ -143,6 +143,7 @@ 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 ae702515b..0895c3a23 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 specifc org in the network")
+	APIEndpointsGetNetworkOrg                   = ffm("api.endpoints.getNetworkOrg", "Gets information about a specific 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,6 +162,7 @@ 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 1921479a5..2f988d58c 100644
--- a/internal/coremsgs/en_struct_descriptions.go
+++ b/internal/coremsgs/en_struct_descriptions.go
@@ -452,16 +452,18 @@ 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")
+	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")
 
 	// 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 6d3b84fa6..8ade6c51e 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -62,6 +62,7 @@ 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()
 
@@ -279,3 +280,7 @@ 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 c6c1c42ff..0e4f64d73 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -108,6 +108,7 @@ 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)
@@ -524,3 +525,19 @@ 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 723f750ab..a2fa147df 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, 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)
+// 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)
 
 	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, remtoeNamespace, signingKey, batch, location)
+		r0 = rf(ctx, nsOpID, networkNamespace, signingKey, batch, location)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index 2689ff43d..4dc55fb28 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -219,6 +219,11 @@ 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 9cfcf99ae..35e9c8cca 100644
--- a/mocks/orchestratormocks/orchestrator.go
+++ b/mocks/orchestratormocks/orchestrator.go
@@ -1223,6 +1223,29 @@ 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 62c8d23c0..b5bcc2c02 100644
--- a/pkg/core/pin.go
+++ b/pkg/core/pin.go
@@ -54,3 +54,8 @@ 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/internal/apiserver/route_post_pins_rewind_test.go b/internal/apiserver/route_post_pins_rewind_test.go
new file mode 100644
index 000000000..2fbbc3379
--- /dev/null
+++ b/internal/apiserver/route_post_pins_rewind_test.go
@@ -0,0 +1,45 @@
+// 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 4a53c871f..23c5e5943 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) {
 
-	ag, cancel := newTestAggregator()
-	defer cancel()
+	em, done := newTestEventManager(t)
+	defer done()
 
-	ag.rewinder.queuedRewinds = []*rewind{{rewindType: rewindBatch}}
-	batchIDs := ag.rewinder.popRewinds()
+	em.QueueBatchRewind(fftypes.NewUUID())
+	batchIDs := em.aggregator.rewinder.popRewinds()
 	assert.Empty(t, batchIDs)
 
-	ag.rewinder.queuedRewinds = []*rewind{{rewindType: rewindBatch}}
-	batchIDs = ag.rewinder.popRewinds()
+	em.QueueBatchRewind(fftypes.NewUUID())
+	batchIDs = em.aggregator.rewinder.popRewinds()
 	assert.Empty(t, batchIDs)
 
 }
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index f4d6c11f7..61ed0815b 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -477,3 +477,54 @@ 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)
+}