Repository: hyperledger/firefly (go)

Issue:
New Orgs / Nodes Rejecting Previous Node Identity Claims
## Description

When creating multi-org networks via the Helm chart, we would stand up the first node for an org, then stand up a new node / org, then another and so on (a bit different from how the CLI creates networks). In the new nodes, we would observe them rejecting the first org's node identity claim messages due to:

```
[2022-04-11T17:19:21.074Z]  WARN poc4-par2-firefly: Unable to process identity claim 564be456-7834-4218-84b9-1a82c9011270: FF10214: Identity '8e8b6544-df58-43fe-935e-f0078b760e7d' not found in identity chain for did:firefly:node/poc4-par1-firefly 'd81e6067-bb73-4613-83c3-49b4a0a08051' dbtx=85C5Fddx role=aggregator
```

The ethconnects were in catchup mode, so the other nodes were presumed to be processing the org and node identity claim messages in the same batchpin. So it appears to be a bug in the logic where the aggregator does not account for needing to process org identity claims before node identity claims within the same batch. 

## Other Info

FireFly Version: `1.0.0-rc3`
Ethconnect Version: `v3.1.5`
Blockchain Provider: `Quorum v21.10.0 (Geth v1.9.25)`

Relevant Ethconnect Configuration:

```yaml
rest:
  rest-gateway:
    leveldb:
      path: "/qdata/ethconnect/receiptsdb"
      queryLimit: 100
    alwaysManageNonce: false
    maxTXWaitTime: 120
    openapi:
      storagePath: "/qdata/ethconnect/openapi"
      eventsDB: "/qdata/ethconnect/eventsdb"
      catchupModeBlockGap: 10000
      catchupModePageSize: 10000
```

## Logs

First Org / Node : [poc4-par1.log](https://github.com/hyperledger/firefly/files/8466801/poc4-par1.log)

Other Orgs / Nodes:
* [poc4-par2.log](https://github.com/hyperledger/firefly/files/8466802/poc4-par2.log)
* [poc4-par3.log](https://github.com/hyperledger/firefly/files/8466803/poc4-par3.log)

Pull request:
DID rewind
Fixes #692 
In PR chain with #707

This PR adds a new reason for a rewind - to go back and confirm messages that were previously parked because the author was not known.

We cannot assure that the identity verification for an identity, will be processed by the aggregator before messages sent by that identity, or child identities registered under that identity. Because they are all sent on different topics.

Patch:
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index a2ab48eec..03a340b18 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -13,11 +13,13 @@ jobs:
     strategy:
       matrix:
         test-suite: [TestEthereumE2ESuite, TestFabricE2ESuite]
-        blockchain-provider: [geth, fabric]
+        blockchain-provider: [geth, besu, fabric]
         database-type: [sqlite3, postgres]
         exclude:
           - blockchain-provider: geth
             test-suite: TestFabricE2ESuite
+          - blockchain-provider: besu
+            test-suite: TestFabricE2ESuite
           - blockchain-provider: fabric
             test-suite: TestEthereumE2ESuite
       fail-fast: false
diff --git a/docs/reference/config.md b/docs/reference/config.md
index 55c9c8bd2..ca7f4904b 100644
--- a/docs/reference/config.md
+++ b/docs/reference/config.md
@@ -411,6 +411,7 @@ nav_order: 3
 |batchTimeout|How long to wait for new events to arrive before performing aggregation on a page of events|[`time.Duration`](https://pkg.go.dev/time#Duration)|`<nil>`
 |firstEvent|The first event the aggregator should process, if no previous offest is stored in the DB. Valid options are `oldest` or `newest`|`string`|`<nil>`
 |pollTimeout|The time to wait without a notification of new events, before trying a select on the table|[`time.Duration`](https://pkg.go.dev/time#Duration)|`<nil>`
+|rewindQueryLimit|Safety limit on the maximum number of records to search when performing queries to search for rewinds|`int`|`<nil>`
 |rewindQueueLength|The size of the queue into the rewind dispatcher|`int`|`<nil>`
 |rewindTimeout|The minimum time to wait for rewinds to accumulate before resolving them|[`time.Duration`](https://pkg.go.dev/time#Duration)|`<nil>`
 
diff --git a/internal/coreconfig/coreconfig.go b/internal/coreconfig/coreconfig.go
index b04b70fae..27d6d853a 100644
--- a/internal/coreconfig/coreconfig.go
+++ b/internal/coreconfig/coreconfig.go
@@ -151,6 +151,8 @@ var (
 	EventAggregatorRewindTimeout = ffc("event.aggregator.rewindTimeout")
 	// EventAggregatorRewindQueueLength the size of the queue into the rewind dispatcher
 	EventAggregatorRewindQueueLength = ffc("event.aggregator.rewindQueueLength")
+	// EventAggregatorRewindQueryLimit safety limit on the maximum number of records to search when performing queries to search for rewinds
+	EventAggregatorRewindQueryLimit = ffc("event.aggregator.rewindQueryLimit")
 	// EventAggregatorRetryFactor the backoff factor to use for retry of database operations
 	EventAggregatorRetryFactor = ffc("event.aggregator.retry.factor")
 	// EventAggregatorRetryInitDelay the initial delay to use for retry of data base operations
@@ -324,6 +326,7 @@ func setDefaults() {
 	viper.SetDefault(string(EventAggregatorPollTimeout), "30s")
 	viper.SetDefault(string(EventAggregatorRewindTimeout), "50ms")
 	viper.SetDefault(string(EventAggregatorRewindQueueLength), 10)
+	viper.SetDefault(string(EventAggregatorRewindQueryLimit), 1000)
 	viper.SetDefault(string(EventAggregatorRetryFactor), 2.0)
 	viper.SetDefault(string(EventAggregatorRetryInitDelay), "100ms")
 	viper.SetDefault(string(EventAggregatorRetryMaxDelay), "30s")
diff --git a/internal/coremsgs/en_config_descriptions.go b/internal/coremsgs/en_config_descriptions.go
index 33a8f69fd..3f5ed5c77 100644
--- a/internal/coremsgs/en_config_descriptions.go
+++ b/internal/coremsgs/en_config_descriptions.go
@@ -176,8 +176,8 @@ var (
 	ConfigEventAggregatorPollTimeout       = ffc("config.event.aggregator.pollTimeout", "The time to wait without a notification of new events, before trying a select on the table", i18n.TimeDurationType)
 	ConfigEventAggregatorRewindQueueLength = ffc("config.event.aggregator.rewindQueueLength", "The size of the queue into the rewind dispatcher", i18n.IntType)
 	ConfigEventAggregatorRewindTimout      = ffc("config.event.aggregator.rewindTimeout", "The minimum time to wait for rewinds to accumulate before resolving them", i18n.TimeDurationType)
-
-	ConfigEventDbeventsBufferSize = ffc("config.event.dbevents.bufferSize", "The size of the buffer of change events", i18n.ByteSizeType)
+	ConfigEventAggregatorRewindQueryLimit  = ffc("config.event.aggregator.rewindQueryLimit", "Safety limit on the maximum number of records to search when performing queries to search for rewinds", i18n.IntType)
+	ConfigEventDbeventsBufferSize          = ffc("config.event.dbevents.bufferSize", "The size of the buffer of change events", i18n.ByteSizeType)
 
 	ConfigEventDispatcherBatchTimeout = ffc("config.event.dispatcher.batchTimeout", "A short time to wait for new events to arrive before re-polling for new events", i18n.TimeDurationType)
 	ConfigEventDispatcherBufferLength = ffc("config.event.dispatcher.bufferLength", "The number of events + attachments an individual dispatcher should hold in memory ready for delivery to the subscription", i18n.IntType)
diff --git a/internal/database/sqlcommon/data_sql.go b/internal/database/sqlcommon/data_sql.go
index 8147bb33c..d46449b87 100644
--- a/internal/database/sqlcommon/data_sql.go
+++ b/internal/database/sqlcommon/data_sql.go
@@ -143,7 +143,7 @@ func (s *SQLCommon) UpsertData(ctx context.Context, data *fftypes.Data, optimiza
 	}
 
 	if !optimized {
-		// Do a select within the transaction to detemine if the UUID already exists
+		// Do a select within the transaction to determine if the UUID already exists
 		dataRows, _, err := s.queryTx(ctx, tx,
 			sq.Select("hash").
 				From("data").
diff --git a/internal/definitions/definition_handler.go b/internal/definitions/definition_handler.go
index 956cb81aa..1dae3749b 100644
--- a/internal/definitions/definition_handler.go
+++ b/internal/definitions/definition_handler.go
@@ -91,6 +91,9 @@ type DefinitionBatchState interface {
 
 	// GetPendingConfirm returns a map of messages are that pending confirmation after already being processed in this batch
 	GetPendingConfirm() map[fftypes.UUID]*fftypes.Message
+
+	// Notify of a DID claim locking in, so a rewind gets queued for it to go back and process any dependent child identities/messages
+	DIDClaimConfirmed(did string)
 }
 
 type definitionHandlers struct {
diff --git a/internal/definitions/definition_handler_identity_claim.go b/internal/definitions/definition_handler_identity_claim.go
index e3aed4565..c1f21a2d0 100644
--- a/internal/definitions/definition_handler_identity_claim.go
+++ b/internal/definitions/definition_handler_identity_claim.go
@@ -134,12 +134,12 @@ func (dh *definitionHandlers) handleIdentityClaim(ctx context.Context, state Def
 	if err != nil && retryable {
 		return HandlerResult{Action: ActionRetry}, err
 	} else if err != nil {
-		// This cannot be processed as the parent does not exist (or similar).
-		// We treat this as a bad request, as nodes should not be broadcast until the parent identity is
-		// is already confirmed. (Note different processing for org/custom child's, where there's a parent
-		// verification to coordinate).
-		l.Warnf("Unable to process identity claim %s: %s", msg.Header.ID, err)
-		return HandlerResult{Action: ActionReject}, nil
+		// This cannot be processed as something in the identity chain is invalid.
+		// We treat this as a park - because we don't know if the parent identity
+		// will be processed after this message and generate a rewind.
+		// They are on separate topics, so there is not ordering assurance between the two messages.
+		l.Infof("Unable to process identity claim (parked) %s: %s", msg.Header.ID, err)
+		return HandlerResult{Action: ActionWait}, nil
 	}
 
 	// Check signature verification
@@ -184,9 +184,10 @@ func (dh *definitionHandlers) handleIdentityClaim(ctx context.Context, state Def
 		if verificationID == nil {
 			// Ok, we still confirm the message as it's valid, and we do not want to block the context.
 			// But we do NOT go on to create the identity - we will be called back
+			log.L(ctx).Infof("Identity %s (%s) awaiting verification claim='%s'", identity.DID, identity.ID, msg.Header.ID)
 			return HandlerResult{Action: ActionConfirm}, nil
 		}
-		log.L(ctx).Infof("Identity '%s' verified claim='%s' verification='%s'", identity.ID, msg.Header.ID, verificationID)
+		log.L(ctx).Infof("Identity %s (%s) verified claim='%s' verification='%s'", identity.DID, identity.ID, msg.Header.ID, verificationID)
 		identity.Messages.Verification = verificationID
 	}
 
@@ -210,6 +211,7 @@ func (dh *definitionHandlers) handleIdentityClaim(ctx context.Context, state Def
 			})
 	}
 
+	state.DIDClaimConfirmed(identity.DID)
 	state.AddFinalize(func(ctx context.Context) error {
 		event := fftypes.NewEvent(fftypes.EventTypeIdentityConfirmed, identity.Namespace, identity.ID, nil, fftypes.SystemTopicDefinitions)
 		return dh.database.InsertEvent(ctx, event)
diff --git a/internal/events/aggregator.go b/internal/events/aggregator.go
index 9667e0917..fb7ac76b0 100644
--- a/internal/events/aggregator.go
+++ b/internal/events/aggregator.go
@@ -135,6 +135,14 @@ func (ag *aggregator) queueBlobRewind(hash *fftypes.Bytes32) {
 	}
 }
 
+func (ag *aggregator) queueDIDRewind(did string) {
+	log.L(ag.ctx).Debugf("Queuing rewind for author DID %s", did)
+	ag.rewinder.rewindRequests <- rewind{
+		rewindType: rewindDIDConfirmed,
+		did:        did,
+	}
+}
+
 func (ag *aggregator) rewindOffchainBatches() (bool, int64) {
 
 	batchIDs := ag.rewinder.popRewinds()
@@ -183,16 +191,19 @@ func (ag *aggregator) processWithBatchState(callback func(ctx context.Context, s
 		return err
 	}
 
-	if len(state.PreFinalize) == 0 {
-		return err
-	}
-
-	if err := state.RunPreFinalize(ag.ctx); err != nil {
-		return err
+	if len(state.PreFinalize) > 0 {
+		if err := state.RunPreFinalize(ag.ctx); err != nil {
+			return err
+		}
+		err = ag.database.RunAsGroup(ag.ctx, func(ctx context.Context) error {
+			return state.RunFinalize(ctx)
+		})
+		if err != nil {
+			return err
+		}
 	}
-	return ag.database.RunAsGroup(ag.ctx, func(ctx context.Context) error {
-		return state.RunFinalize(ctx)
-	})
+	state.queueRewinds(ag)
+	return nil
 }
 
 func (ag *aggregator) processPinsEventsHandler(items []fftypes.LocallySequenced) (repoll bool, err error) {
diff --git a/internal/events/aggregator_batch_state.go b/internal/events/aggregator_batch_state.go
index 4ef8137cd..28e785d7e 100644
--- a/internal/events/aggregator_batch_state.go
+++ b/internal/events/aggregator_batch_state.go
@@ -101,6 +101,7 @@ type batchState struct {
 	unmaskedContexts   map[fftypes.Bytes32]*contextState
 	dispatchedMessages []*dispatchedMessage
 	pendingConfirms    map[fftypes.UUID]*fftypes.Message
+	confirmedDIDClaims []string
 
 	// PreFinalize callbacks may perform blocking actions (possibly to an external connector)
 	// - Will execute after all batch messages have been processed
@@ -149,6 +150,16 @@ func (bs *batchState) RunFinalize(ctx context.Context) error {
 	return bs.flushPins(ctx)
 }
 
+func (bs *batchState) DIDClaimConfirmed(did string) {
+	bs.confirmedDIDClaims = append(bs.confirmedDIDClaims, did)
+}
+
+func (bs *batchState) queueRewinds(ag *aggregator) {
+	for _, did := range bs.confirmedDIDClaims {
+		ag.queueDIDRewind(did)
+	}
+}
+
 func (bs *batchState) CheckUnmaskedContextReady(ctx context.Context, contextUnmasked *fftypes.Bytes32, msg *fftypes.Message, topic string, firstMsgPinSequence int64) (bool, error) {
 
 	ucs, found := bs.unmaskedContexts[*contextUnmasked]
diff --git a/internal/events/aggregator_rewind.go b/internal/events/aggregator_rewind.go
index 92ff3adde..7690ce0bc 100644
--- a/internal/events/aggregator_rewind.go
+++ b/internal/events/aggregator_rewind.go
@@ -37,12 +37,14 @@ const (
 	rewindBatch rewindType = iota
 	rewindMessage
 	rewindBlob
+	rewindDIDConfirmed
 )
 
 type rewind struct {
 	rewindType rewindType
 	uuid       fftypes.UUID
 	hash       fftypes.Bytes32
+	did        string
 }
 
 type rewinder struct {
@@ -60,6 +62,7 @@ type rewinder struct {
 	stagedRewinds    []*rewind
 	loop2ShoulderTap chan bool
 	readyRewinds     map[fftypes.UUID]bool
+	querySafetyLimit uint64
 }
 
 func newRewinder(ag *aggregator) *rewinder {
@@ -74,6 +77,7 @@ func newRewinder(ag *aggregator) *rewinder {
 		rewindRequests:   make(chan rewind, config.GetInt(coreconfig.EventAggregatorRewindQueueLength)),
 		loop2ShoulderTap: make(chan bool, 1),
 		minRewindTimeout: config.GetDuration(coreconfig.EventAggregatorRewindTimeout),
+		querySafetyLimit: uint64(config.GetUint((coreconfig.EventAggregatorRewindQueryLimit))),
 		readyRewinds:     make(map[fftypes.UUID]bool),
 	}
 }
@@ -152,6 +156,7 @@ func (rw *rewinder) processStagedRewinds() bool {
 	batchIDs := make(map[fftypes.UUID]bool)
 	var msgRewinds []*fftypes.UUID
 	var newBlobHashes []driver.Value
+	var identityRewinds []driver.Value
 
 	// Pop the current batch of rewinds out of the staging area
 	rw.mux.Lock()
@@ -163,6 +168,8 @@ func (rw *rewinder) processStagedRewinds() bool {
 			newBlobHashes = append(newBlobHashes, &rewind.hash)
 		case rewindMessage:
 			msgRewinds = append(msgRewinds, &rewind.uuid)
+		case rewindDIDConfirmed:
+			identityRewinds = append(identityRewinds, rewind.did)
 		}
 	}
 	rw.stagedRewinds = rw.stagedRewinds[:0] // truncate
@@ -181,6 +188,11 @@ func (rw *rewinder) processStagedRewinds() bool {
 					return err
 				}
 			}
+			if len(identityRewinds) > 0 {
+				if err := rw.getRewindsForDIDs(ctx, identityRewinds, batchIDs); err != nil {
+					return err
+				}
+			}
 			return nil
 		})
 	})
@@ -257,7 +269,7 @@ func (rw *rewinder) getRewindsForBlobs(ctx context.Context, newHashes []driver.V
 
 	// Find any data associated with this blob
 	var data []*fftypes.DataRef
-	filter := database.DataQueryFactory.NewFilter(ctx).In("blob.hash", newHashes)
+	filter := database.DataQueryFactory.NewFilterLimit(ctx, rw.querySafetyLimit).In("blob.hash", newHashes)
 	data, _, err := rw.database.GetDataRefs(ctx, filter)
 	if err != nil {
 		return err
@@ -283,3 +295,27 @@ func (rw *rewinder) getRewindsForBlobs(ctx context.Context, newHashes []driver.V
 
 	return nil
 }
+
+func (rw *rewinder) getRewindsForDIDs(ctx context.Context, dids []driver.Value, batchIDs map[fftypes.UUID]bool) error {
+
+	// We need to find all pending messages, that are authored by this DID
+	fb := database.MessageQueryFactory.NewFilterLimit(ctx, rw.querySafetyLimit)
+	filter := fb.And(
+		fb.Eq("state", fftypes.MessageStatePending),
+		fb.In("author", dids),
+	)
+	records, err := rw.database.GetMessageIDs(ctx, filter)
+	if err != nil {
+		return err
+	}
+	if len(records) == 0 {
+		return nil
+	}
+	msgIDs := make([]*fftypes.UUID, len(records))
+	for i, record := range records {
+		msgIDs[i] = &record.ID
+	}
+
+	// We can treat the message level rewinds just like any other message rewind
+	return rw.getRewindsForMessages(ctx, msgIDs, batchIDs)
+}
diff --git a/manifest.json b/manifest.json
index 10300c389..66beaa348 100644
--- a/manifest.json
+++ b/manifest.json
@@ -44,6 +44,6 @@
     "release": "v0.7.0_50810a4"
   },
   "cli": {
-    "tag": "v0.0.46"
+    "tag": "v0.0.48"
   }
 }


Test patch:
diff --git a/internal/definitions/definition_handler_identity_claim_test.go b/internal/definitions/definition_handler_identity_claim_test.go
index 4637e1414..981a13a37 100644
--- a/internal/definitions/definition_handler_identity_claim_test.go
+++ b/internal/definitions/definition_handler_identity_claim_test.go
@@ -177,13 +177,14 @@ func TestHandleDefinitionIdentityClaimCustomWithExistingParentVerificationOk(t *
 	action, err := dh.HandleDefinitionBroadcast(ctx, bs, claimMsg, fftypes.DataArray{claimData}, fftypes.NewUUID())
 	assert.Equal(t, HandlerResult{Action: ActionConfirm}, action)
 	assert.NoError(t, err)
+	assert.Equal(t, bs.confirmedDIDClaims, []string{custom1.DID})
 
 	err = bs.finalizers[0](ctx)
 	assert.NoError(t, err)
 
-	mim.AssertExpectations(t)
 	mdi.AssertExpectations(t)
 	mdm.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
@@ -563,7 +564,7 @@ func TestHandleDefinitionIdentityVerifyChainInvalid(t *testing.T) {
 	mim.On("VerifyIdentityChain", ctx, custom1).Return(nil, false, fmt.Errorf("wrong"))
 
 	action, err := dh.HandleDefinitionBroadcast(ctx, bs, claimMsg, fftypes.DataArray{claimData}, fftypes.NewUUID())
-	assert.Equal(t, HandlerResult{Action: ActionReject}, action)
+	assert.Equal(t, HandlerResult{Action: ActionWait}, action)
 	assert.NoError(t, err)
 
 	mim.AssertExpectations(t)
diff --git a/internal/definitions/definition_handler_identity_verification_test.go b/internal/definitions/definition_handler_identity_verification_test.go
index 76b4c7824..fbd718d5a 100644
--- a/internal/definitions/definition_handler_identity_verification_test.go
+++ b/internal/definitions/definition_handler_identity_verification_test.go
@@ -69,6 +69,7 @@ func TestHandleDefinitionIdentityVerificationWithExistingClaimOk(t *testing.T) {
 	action, err := dh.HandleDefinitionBroadcast(ctx, bs, verifyMsg, fftypes.DataArray{verifyData}, fftypes.NewUUID())
 	assert.Equal(t, HandlerResult{Action: ActionConfirm}, action)
 	assert.NoError(t, err)
+	assert.Equal(t, bs.confirmedDIDClaims, []string{custom1.DID})
 
 	err = bs.finalizers[0](ctx)
 	assert.NoError(t, err)
diff --git a/internal/definitions/definition_handler_test.go b/internal/definitions/definition_handler_test.go
index 35c7a9627..47cfcbb0f 100644
--- a/internal/definitions/definition_handler_test.go
+++ b/internal/definitions/definition_handler_test.go
@@ -50,10 +50,11 @@ func newTestDefinitionHandlers(t *testing.T) (*definitionHandlers, *testDefiniti
 }
 
 type testDefinitionBatchState struct {
-	t               *testing.T
-	preFinalizers   []func(ctx context.Context) error
-	finalizers      []func(ctx context.Context) error
-	pendingConfirms map[fftypes.UUID]*fftypes.Message
+	t                  *testing.T
+	preFinalizers      []func(ctx context.Context) error
+	finalizers         []func(ctx context.Context) error
+	pendingConfirms    map[fftypes.UUID]*fftypes.Message
+	confirmedDIDClaims []string
 }
 
 func newTestDefinitionBatchState(t *testing.T) *testDefinitionBatchState {
@@ -75,6 +76,10 @@ func (bs *testDefinitionBatchState) GetPendingConfirm() map[fftypes.UUID]*fftype
 	return bs.pendingConfirms
 }
 
+func (bs *testDefinitionBatchState) DIDClaimConfirmed(did string) {
+	bs.confirmedDIDClaims = append(bs.confirmedDIDClaims, did)
+}
+
 func (bs *testDefinitionBatchState) assertNoFinalizers() {
 	assert.Empty(bs.t, bs.preFinalizers)
 	assert.Empty(bs.t, bs.finalizers)
diff --git a/internal/events/aggregator_rewind_test.go b/internal/events/aggregator_rewind_test.go
index 13808386e..427633c79 100644
--- a/internal/events/aggregator_rewind_test.go
+++ b/internal/events/aggregator_rewind_test.go
@@ -37,6 +37,7 @@ func TestRewinderE2E(t *testing.T) {
 	batchID2 := fftypes.NewUUID()
 	batchID3 := fftypes.NewUUID()
 	batchID4 := fftypes.NewUUID()
+	batchID5 := fftypes.NewUUID()
 
 	mdi := ag.database.(*databasemocks.Plugin)
 	mdm := ag.data.(*datamocks.Manager)
@@ -48,14 +49,19 @@ func TestRewinderE2E(t *testing.T) {
 		Return([]*fftypes.UUID{batchID2}, nil)
 	mdm.On("PeekMessageCache", mock.Anything, mock.Anything, data.CRORequireBatchID).Return(nil, nil)
 	mdi.On("GetBatchIDsForMessages", mock.Anything, mock.Anything).
-		Return([]*fftypes.UUID{batchID3}, nil)
+		Return([]*fftypes.UUID{batchID3}, nil).Once()
+	mdi.On("GetMessageIDs", mock.Anything, mock.Anything).
+		Return([]*fftypes.IDAndSequence{{ID: *fftypes.NewUUID()}}, nil).Once()
+	mdi.On("GetBatchIDsForMessages", mock.Anything, mock.Anything).
+		Return([]*fftypes.UUID{batchID4}, nil).Once()
 
 	ag.rewinder.start()
 
 	ag.queueBatchRewind(batchID1)
 	ag.queueBlobRewind(fftypes.NewRandB32())
 	ag.queueMessageRewind(fftypes.NewUUID())
-	ag.queueBatchRewind(batchID4)
+	ag.queueDIDRewind("did:firefly:org/bob")
+	ag.queueBatchRewind(batchID5)
 
 	allReceived := make(map[fftypes.UUID]bool)
 	for len(allReceived) < 4 {
@@ -69,6 +75,7 @@ func TestRewinderE2E(t *testing.T) {
 	assert.True(t, allReceived[*batchID2])
 	assert.True(t, allReceived[*batchID3])
 	assert.True(t, allReceived[*batchID4])
+	assert.True(t, allReceived[*batchID5])
 
 	cancel()
 	<-ag.rewinder.loop1Done
@@ -78,7 +85,7 @@ func TestRewinderE2E(t *testing.T) {
 	mdi.AssertExpectations(t)
 }
 
-func TestPrcessStagedRewindsErrorMessages(t *testing.T) {
+func TestProcessStagedRewindsErrorMessages(t *testing.T) {
 
 	ag, cancel := newTestAggregator()
 	cancel()
@@ -100,7 +107,7 @@ func TestPrcessStagedRewindsErrorMessages(t *testing.T) {
 
 }
 
-func TestPrcessStagedRewindsMessagesCached(t *testing.T) {
+func TestProcessStagedRewindsMessagesCached(t *testing.T) {
 
 	ag, cancel := newTestAggregator()
 	cancel()
@@ -123,7 +130,7 @@ func TestPrcessStagedRewindsMessagesCached(t *testing.T) {
 
 }
 
-func TestPrcessStagedRewindsErrorBlobBatchIDs(t *testing.T) {
+func TestProcessStagedRewindsErrorBlobBatchIDs(t *testing.T) {
 
 	ag, cancel := newTestAggregator()
 	cancel()
@@ -147,7 +154,7 @@ func TestPrcessStagedRewindsErrorBlobBatchIDs(t *testing.T) {
 
 }
 
-func TestPrcessStagedRewindsErrorBlobDataRefs(t *testing.T) {
+func TestProcessStagedRewindsErrorBlobDataRefs(t *testing.T) {
 
 	ag, cancel := newTestAggregator()
 	cancel()
@@ -167,6 +174,46 @@ func TestPrcessStagedRewindsErrorBlobDataRefs(t *testing.T) {
 
 }
 
+func TestProcessStagedRewindsErrorDIDs(t *testing.T) {
+
+	ag, cancel := newTestAggregator()
+	cancel()
+
+	mdi := ag.database.(*databasemocks.Plugin)
+
+	mockRunAsGroupPassthrough(mdi)
+	mdi.On("GetMessageIDs", mock.Anything, mock.Anything).
+		Return(nil, fmt.Errorf("pop"))
+
+	ag.rewinder.stagedRewinds = []*rewind{
+		{rewindType: rewindDIDConfirmed},
+	}
+	ag.rewinder.processStagedRewinds()
+
+	mdi.AssertExpectations(t)
+
+}
+
+func TestProcessStagedRewindsNoDIDs(t *testing.T) {
+
+	ag, cancel := newTestAggregator()
+	cancel()
+
+	mdi := ag.database.(*databasemocks.Plugin)
+
+	mockRunAsGroupPassthrough(mdi)
+	mdi.On("GetMessageIDs", mock.Anything, mock.Anything).
+		Return([]*fftypes.IDAndSequence{}, nil)
+
+	ag.rewinder.stagedRewinds = []*rewind{
+		{rewindType: rewindDIDConfirmed},
+	}
+	ag.rewinder.processStagedRewinds()
+
+	mdi.AssertExpectations(t)
+
+}
+
 func TestPopRewindsDoublePopNoBlock(t *testing.T) {
 
 	ag, cancel := newTestAggregator()
diff --git a/internal/events/aggregator_test.go b/internal/events/aggregator_test.go
index a32cb931c..55ef5ac4f 100644
--- a/internal/events/aggregator_test.go
+++ b/internal/events/aggregator_test.go
@@ -1774,7 +1774,7 @@ func TestDefinitionBroadcastRejectBadSigner(t *testing.T) {
 
 }
 
-func TestDefinitionBroadcastRejectUnregisteredSignerIdentityClaim(t *testing.T) {
+func TestDefinitionBroadcastParkUnregisteredSignerIdentityClaim(t *testing.T) {
 	ag, cancel := newTestAggregator()
 	defer cancel()
 
@@ -1787,9 +1787,10 @@ func TestDefinitionBroadcastRejectUnregisteredSignerIdentityClaim(t *testing.T)
 	msh := ag.definitions.(*definitionsmocks.DefinitionHandlers)
 	msh.On("HandleDefinitionBroadcast", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(definitions.HandlerResult{Action: definitions.ActionWait}, nil)
 
-	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &fftypes.Pin{Signer: "0x12345"})
+	newState, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &fftypes.Pin{Signer: "0x12345"})
 	assert.NoError(t, err)
 	assert.False(t, valid)
+	assert.Empty(t, newState)
 
 	mim.AssertExpectations(t)
 	msh.AssertExpectations(t)
@@ -2104,6 +2105,41 @@ func TestProcessWithBatchActionsSuccess(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestProcessWithBatchRewindsSuccess(t *testing.T) {
+	ag, cancel := newTestAggregator()
+	defer cancel()
+
+	mdi := ag.database.(*databasemocks.Plugin)
+	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything).Maybe()
+	rag.RunFn = func(a mock.Arguments) {
+		rag.ReturnArguments = mock.Arguments{a[1].(func(context.Context) error)(a[0].(context.Context))}
+	}
+
+	err := ag.processWithBatchState(func(ctx context.Context, actions *batchState) error {
+		actions.DIDClaimConfirmed("did:firefly:org/test")
+		return nil
+	})
+	assert.NoError(t, err)
+}
+
+func TestProcessWithBatchActionsFail(t *testing.T) {
+	ag, cancel := newTestAggregator()
+	defer cancel()
+
+	mdi := ag.database.(*databasemocks.Plugin)
+	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything).Once()
+	rag.RunFn = func(a mock.Arguments) {
+		rag.ReturnArguments = mock.Arguments{a[1].(func(context.Context) error)(a[0].(context.Context))}
+	}
+	mdi.On("RunAsGroup", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+
+	err := ag.processWithBatchState(func(ctx context.Context, actions *batchState) error {
+		actions.AddPreFinalize(func(ctx context.Context) error { return nil })
+		return nil
+	})
+	assert.EqualError(t, err, "pop")
+}
+
 func TestExtractManifestFail(t *testing.T) {
 	ag, cancel := newTestAggregator()
 	defer cancel()
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index fc8f10593..45df009b1 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -37,18 +37,20 @@ import (
 )
 
 type testState struct {
-	startTime time.Time
-	t         *testing.T
-	client1   *resty.Client
-	client2   *resty.Client
-	ethNode   *resty.Client
-	ws1       *websocket.Conn
-	ws2       *websocket.Conn
-	org1      *fftypes.Identity
-	org1key   *fftypes.Verifier
-	org2      *fftypes.Identity
-	org2key   *fftypes.Verifier
-	done      func()
+	startTime            time.Time
+	t                    *testing.T
+	client1              *resty.Client
+	client2              *resty.Client
+	ethNode              *resty.Client
+	ws1                  *websocket.Conn
+	ws2                  *websocket.Conn
+	org1                 *fftypes.Identity
+	org1key              *fftypes.Verifier
+	org2                 *fftypes.Identity
+	org2key              *fftypes.Verifier
+	done                 func()
+	stackState           *StackState
+	unregisteredAccounts []interface{}
 }
 
 var widgetSchemaJSON = []byte(`{
@@ -163,17 +165,30 @@ func readStackFile(t *testing.T) *Stack {
 	return stack
 }
 
+func readStackState(t *testing.T) *StackState {
+	stackFile := os.Getenv("STACK_STATE")
+	if stackFile == "" {
+		t.Fatal("STACK_STATE must be set")
+	}
+	stackState, err := ReadStackState(stackFile)
+	assert.NoError(t, err)
+	return stackState
+}
+
 func beforeE2ETest(t *testing.T) *testState {
 	stack := readStackFile(t)
+	stackState := readStackState(t)
 
 	var authHeader1 http.Header
 	var authHeader2 http.Header
 
 	ts := &testState{
-		t:         t,
-		startTime: time.Now(),
-		client1:   NewResty(t),
-		client2:   NewResty(t),
+		t:                    t,
+		startTime:            time.Now(),
+		client1:              NewResty(t),
+		client2:              NewResty(t),
+		stackState:           stackState,
+		unregisteredAccounts: stackState.Accounts[2:],
 	}
 
 	httpProtocolClient1 := "http"
diff --git a/test/e2e/fabric_test.go b/test/e2e/fabric_test.go
index 48ad3f85b..9970f651d 100644
--- a/test/e2e/fabric_test.go
+++ b/test/e2e/fabric_test.go
@@ -23,6 +23,7 @@ import (
 )
 
 func TestFabricE2ESuite(t *testing.T) {
+	suite.Run(t, new(IdentityTestSuite))
 	suite.Run(t, new(OnChainOffChainTestSuite))
 	suite.Run(t, new(FabricContractTestSuite))
 }
diff --git a/test/e2e/identity_test.go b/test/e2e/identity_test.go
index 1c5e7e6ee..6d1167d09 100644
--- a/test/e2e/identity_test.go
+++ b/test/e2e/identity_test.go
@@ -19,6 +19,7 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/hyperledger/firefly/pkg/fftypes"
@@ -43,18 +44,13 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityBroadcasts() {
 	received1 := wsReader(suite.testState.ws1, false)
 	received2 := wsReader(suite.testState.ws2, false)
 
-	// Create some keys
-	totalIdentities := 3
-	keys := make([]string, totalIdentities)
-	for i := 0; i < totalIdentities; i++ {
-		keys[i] = CreateEthAccount(suite.T(), suite.testState.ethNode)
-	}
-
+	totalIdentities := 2
 	ts := time.Now().Unix()
 	for i := 0; i < totalIdentities; i++ {
+		key := getUnregisteredAccount(suite, suite.testState.org1.Name)
 		ClaimCustomIdentity(suite.T(),
 			suite.testState.client1,
-			keys[i],
+			key,
 			fmt.Sprintf("custom_%d_%d", ts, i),
 			fmt.Sprintf("Description %d", i),
 			fftypes.JSONObject{"profile": i},
@@ -66,8 +62,10 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityBroadcasts() {
 	for i := 0; i < totalIdentities; i++ {
 		ed := waitForIdentityConfirmed(suite.T(), received1)
 		identityIDs[*ed.Reference] = true
+		suite.T().Logf("Received node 1 confirmation of identity %s", ed.Reference)
 		ed = waitForIdentityConfirmed(suite.T(), received2)
 		identityIDs[*ed.Reference] = true
+		suite.T().Logf("Received node 2 confirmation of identity %s", ed.Reference)
 	}
 	assert.Len(suite.T(), identityIDs, totalIdentities)
 
@@ -100,14 +98,13 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityPrivate() {
 	received1 := wsReader(suite.testState.ws1, false)
 	received2 := wsReader(suite.testState.ws2, false)
 
-	// Create an identity on both sides
-	org1Key := CreateEthAccount(suite.T(), suite.testState.ethNode)
-	org2Key := CreateEthAccount(suite.T(), suite.testState.ethNode)
+	org1key := getUnregisteredAccount(suite, suite.testState.org1.Name)
+	org2key := getUnregisteredAccount(suite, suite.testState.org2.Name)
 
 	ts := time.Now().Unix()
 	custom1 := ClaimCustomIdentity(suite.T(),
 		suite.testState.client1,
-		org1Key,
+		org1key,
 		fmt.Sprintf("custom_%d_org1priv", ts),
 		fmt.Sprintf("Description org1priv"),
 		nil,
@@ -115,7 +112,7 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityPrivate() {
 		true)
 	custom2 := ClaimCustomIdentity(suite.T(),
 		suite.testState.client2,
-		org2Key,
+		org2key,
 		fmt.Sprintf("custom_%d_org2priv", ts),
 		fmt.Sprintf("Description org2priv"),
 		nil,
@@ -126,7 +123,7 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityPrivate() {
 		waitForIdentityConfirmed(suite.T(), received2)
 	}
 
-	resp, err := PrivateMessageWithKey(suite.testState, suite.testState.client1, org1Key, "topic1", &fftypes.DataRefOrValue{
+	resp, err := PrivateMessageWithKey(suite.testState, suite.testState.client1, org1key, "topic1", &fftypes.DataRefOrValue{
 		Value: fftypes.JSONAnyPtr(`"test private custom identity"`),
 	}, []string{custom1.DID, custom2.DID}, "tag1", fftypes.TransactionTypeBatchPin, true)
 	require.NoError(suite.T(), err)
@@ -134,5 +131,46 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityPrivate() {
 
 	waitForMessageConfirmed(suite.T(), received1, fftypes.MessageTypePrivate)
 	waitForMessageConfirmed(suite.T(), received2, fftypes.MessageTypePrivate)
+}
 
+func getUnregisteredAccount(suite *IdentityTestSuite, orgName string) string {
+	verifiers := GetVerifiers(suite.T(), suite.testState.client1)
+	suite.T().Logf("checking for account with orgName: %s", orgName)
+	for i, account := range suite.testState.unregisteredAccounts {
+		alreadyRegisted := false
+		accountMap := account.(map[string]interface{})
+		suite.T().Logf("account from stackState has orgName: %s", accountMap["orgName"])
+		if accountOrgName, ok := accountMap["orgName"]; ok {
+			if accountOrgName != orgName {
+				// An orgName was present and it wasn't what we were looking for. Skip.
+				continue
+			}
+		}
+		var key string
+		if k, ok := accountMap["address"]; ok {
+			key = k.(string)
+		}
+		if k, ok := accountMap["name"]; ok {
+			key = k.(string)
+		}
+
+		suite.T().Logf("checking to make sure key '%s' is not registered", key)
+		if len(verifiers) > 0 {
+			for _, verifier := range verifiers {
+
+				suite.T().Logf("account name/address: %s", key)
+				suite.T().Logf("verifier value: %s", verifier.Value)
+				if strings.Contains(verifier.Value, key) {
+					// Already registered. Look at the next account
+					alreadyRegisted = true
+					break
+				}
+			}
+		}
+		if !alreadyRegisted {
+			suite.testState.unregisteredAccounts = append(suite.testState.unregisteredAccounts[:i], suite.testState.unregisteredAccounts[i+1:]...)
+			return key
+		}
+	}
+	return ""
 }
diff --git a/test/e2e/restclient_test.go b/test/e2e/restclient_test.go
index 8c3607ade..ae094b48d 100644
--- a/test/e2e/restclient_test.go
+++ b/test/e2e/restclient_test.go
@@ -47,6 +47,7 @@ var (
 	urlDatatypes         = "/namespaces/default/datatypes"
 	urlIdentities        = "/namespaces/default/identities"
 	urlIdentity          = "/namespaces/default/identities/%s"
+	urlVerifiers         = "/namespaces/default/verifiers"
 	urlTokenPools        = "/namespaces/default/tokens/pools"
 	urlTokenMint         = "/namespaces/default/tokens/mint"
 	urlTokenBurn         = "/namespaces/default/tokens/burn"
@@ -235,28 +236,6 @@ func BroadcastMessageAsIdentity(t *testing.T, client *resty.Client, did, topic s
 	return res, err
 }
 
-func CreateEthAccount(t *testing.T, client *resty.Client) string {
-	createPayload := map[string]interface{}{"jsonrpc": "2.0", "id": 0, "method": "personal_newAccount", "params": []interface{}{""}}
-	var resBody struct {
-		Result string `json:"result"`
-	}
-	res, err := client.R().
-		SetBody(createPayload).
-		SetResult(&resBody).
-		Post("/")
-	assert.NoError(t, err)
-	assert.Equal(t, 200, res.StatusCode())
-	newKey := resBody.Result
-	t.Logf("New key: %s", newKey)
-	unlockPayload := map[string]interface{}{"jsonrpc": "2.0", "id": 0, "method": "personal_unlockAccount", "params": []interface{}{newKey, "", 0}}
-	res, err = client.R().
-		SetBody(unlockPayload).
-		Post("/")
-	assert.NoError(t, err)
-	assert.Equal(t, 200, res.StatusCode())
-	return newKey
-}
-
 func ClaimCustomIdentity(t *testing.T, client *resty.Client, key, name, desc string, profile fftypes.JSONObject, parent *fftypes.UUID, confirm bool) *fftypes.Identity {
 	var identity fftypes.Identity
 	res, err := client.R().
@@ -275,6 +254,7 @@ func ClaimCustomIdentity(t *testing.T, client *resty.Client, key, name, desc str
 		Post(urlIdentities)
 	assert.NoError(t, err)
 	assert.True(t, res.IsSuccess())
+	t.Logf("Identity creation initiated with key %s: %s (%s)", key, identity.ID, identity.DID)
 	return &identity
 }
 
@@ -288,6 +268,16 @@ func GetIdentity(t *testing.T, client *resty.Client, id *fftypes.UUID) *fftypes.
 	return &identity
 }
 
+func GetVerifiers(t *testing.T, client *resty.Client) []*fftypes.Verifier {
+	var verifiers []*fftypes.Verifier
+	res, err := client.R().
+		SetResult(&verifiers).
+		Get(urlVerifiers)
+	assert.NoError(t, err)
+	assert.True(t, res.IsSuccess())
+	return verifiers
+}
+
 func CreateBlob(t *testing.T, client *resty.Client, dt *fftypes.DatatypeRef) *fftypes.Data {
 	r, _ := rand.Int(rand.Reader, big.NewInt(1024*1024))
 	blob := make([]byte, r.Int64()+1024*1024)
diff --git a/test/e2e/run.sh b/test/e2e/run.sh
index b259722f7..75bfd43ed 100755
--- a/test/e2e/run.sh
+++ b/test/e2e/run.sh
@@ -7,7 +7,7 @@ set -o pipefail
 CWD=$(dirname "$0")
 CLI="ff -v --ansi never"
 CLI_VERSION=$(cat $CWD/../../manifest.json | jq -r .cli.tag)
-STACK_DIR=~/.firefly/stacks
+STACKS_DIR=~/.firefly/stacks
 
 checkOk() {
   local rc=$1
@@ -44,9 +44,12 @@ if [ -z "${DATABASE_TYPE}" ]; then
 fi
 
 if [ -z "${STACK_FILE}" ]; then
-  STACK_FILE=$STACK_DIR/$STACK_NAME/stack.json
+  STACK_FILE=$STACKS_DIR/$STACK_NAME/stack.json
 fi
 
+if [ -z "${STACK_STATE}" ]; then
+  STACK_STATE=$STACKS_DIR/$STACK_NAME/runtime/stackState.json
+fi
 
 if [ -z "${BLOCKCHAIN_PROVIDER}" ]; then
   BLOCKCHAIN_PROVIDER=geth
@@ -77,7 +80,7 @@ if [ "$DOWNLOAD_CLI" == "true" ]; then
 fi
 
 if [ "$CREATE_STACK" == "true" ]; then
-  $CLI init --prometheus-enabled --database $DATABASE_TYPE $STACK_NAME 2 --blockchain-provider $BLOCKCHAIN_PROVIDER --token-providers $TOKENS_PROVIDER --manifest ../../manifest.json $EXTRA_INIT_ARGS
+  $CLI init --prometheus-enabled --database $DATABASE_TYPE $STACK_NAME 2 --blockchain-provider $BLOCKCHAIN_PROVIDER --token-providers $TOKENS_PROVIDER --manifest ../../manifest.json $EXTRA_INIT_ARGS --sandbox-enabled=false
   checkOk $?
 
   $CLI pull $STACK_NAME -r 3
@@ -87,25 +90,33 @@ if [ "$CREATE_STACK" == "true" ]; then
   checkOk $?
 
   if [ "$TEST_SUITE" == "TestEthereumE2ESuite" ]; then
-      prefix='contract address: '
-      output=$($CLI deploy $STACK_NAME ../data/simplestorage/simple_storage.json | grep address)
-      export CONTRACT_ADDRESS=${output#"$prefix"}
+      export CONTRACT_ADDRESS=$($CLI deploy ethereum $STACK_NAME ../data/simplestorage/simple_storage.json | jq -r '.address')
+      # Create 5 new accounts for use in testing
+      for i in {1..4}
+      do
+          $CLI accounts create $STACK_NAME
+      done
+  elif [ "$TEST_SUITE" == "TestFabricE2ESuite" ]; then
+      # Create 5 new accounts for use in testing
+      for i in {1..3}
+      do
+          $CLI accounts create $STACK_NAME org_0  user_$i
+      done
+      # Create one account that is specifically only usable from the second org
+      $CLI accounts create $STACK_NAME org_1  user_5
   fi
 fi
 
 if [ "$TOKENS_PROVIDER" == "erc20_erc721" ]; then
-    prefix='contract address: '
-    output=$($CLI deploy $STACK_NAME ../data/erc20/ERC20WithData.json | grep address)
-    export ERC20_CONTRACT_ADDRESS=${output#"$prefix"}
-    prefix='contract address: '
-    output=$($CLI deploy $STACK_NAME ../data/erc721/ERC721WithData.json | grep address)
-    export ERC721_CONTRACT_ADDRESS=${output#"$prefix"}
+    export ERC20_CONTRACT_ADDRESS=$($CLI deploy ethereum $STACK_NAME ../data/erc20/ERC20WithData.json | jq -r '.address')
+    export ERC721_CONTRACT_ADDRESS=$($CLI deploy ethereum $STACK_NAME ../data/erc721/ERC721WithData.json | jq -r '.address')
 fi
 
 $CLI info $STACK_NAME
 checkOk $?
 
 export STACK_FILE
+export STACK_STATE
 
 go clean -testcache && go test -v . -run $TEST_SUITE
 checkOk $?
diff --git a/test/e2e/stack.go b/test/e2e/stack.go
index d3004be39..42f5212af 100644
--- a/test/e2e/stack.go
+++ b/test/e2e/stack.go
@@ -23,20 +23,25 @@ import (
 
 type Stack struct {
 	Name                  string    `json:"name,omitempty"`
-	ExposedBlockchainPort int       `json:"exposedGethPort,omitempty"`
+	ExposedBlockchainPort int       `json:"exposedBlockchainPort,omitempty"`
 	BlockchainProvider    string    `json:"blockchainProvider"`
 	TokenProviders        []string  `json:"tokenProviders"`
 	Members               []*Member `json:"members,omitempty"`
 }
 
+type StackState struct {
+	Accounts []interface{} `json:"accounts"`
+}
+
 type Member struct {
-	ExposedFireflyPort   int    `json:"exposedFireflyPort,omitempty"`
-	FireflyHostname      string `json:"fireflyHostname,omitempty"`
-	Username             string `json:"username,omitempty"`
-	Password             string `json:"password,omitempty"`
-	UseHTTPS             bool   `json:"useHttps,omitempty"`
-	ExposedConnectorPort int    `json:"exposedConnectorPort,omitempty"`
-	OrgName              string `json:"orgName,omitempty"`
+	ExposedFireflyPort   int         `json:"exposedFireflyPort,omitempty"`
+	FireflyHostname      string      `json:"fireflyHostname,omitempty"`
+	Username             string      `json:"username,omitempty"`
+	Password             string      `json:"password,omitempty"`
+	UseHTTPS             bool        `json:"useHttps,omitempty"`
+	ExposedConnectorPort int         `json:"exposedConnectorPort,omitempty"`
+	OrgName              string      `json:"orgName,omitempty"`
+	Account              interface{} `json:"account,omitempty"`
 }
 
 func GetMemberPort(filename string, n int) (int, error) {
@@ -91,3 +96,18 @@ func ReadStack(filename string) (*Stack, error) {
 
 	return stack, nil
 }
+
+func ReadStackState(filename string) (*StackState, error) {
+	jsonBytes, err := ioutil.ReadFile(filename)
+	if err != nil {
+		return nil, err
+	}
+
+	var stackState *StackState
+	err = json.Unmarshal(jsonBytes, &stackState)
+	if err != nil {
+		return nil, err
+	}
+
+	return stackState, nil
+}