Add NamespaceInvalidState and NamespaceNotFound errors (#2785)

client/history/metricClient.go

@@ -632,6 +632,7 @@ func (c *metricClient) finishMetricsRecording(
 		case *serviceerror.Canceled,
 			*serviceerror.DeadlineExceeded,
 			*serviceerror.NotFound,
+			*serviceerror.NamespaceNotFound,
 			*serviceerror.WorkflowExecutionAlreadyStarted:
 			// noop - not interest and too many logs
 		default:

client/matching/metricClient.go

@@ -261,6 +261,7 @@ func (c *metricClient) finishMetricsRecording(
 		case *serviceerror.Canceled,
 			*serviceerror.DeadlineExceeded,
 			*serviceerror.NotFound,
+			*serviceerror.NamespaceNotFound,
 			*serviceerror.WorkflowExecutionAlreadyStarted:
 			// noop - not interest and too many logs
 		default:

common/archiver/filestore/historyArchiver.go

@@ -153,7 +153,7 @@ func (h *historyArchiver) Archive(
 	for historyIterator.HasNext() {
 		historyBlob, err := getNextHistoryBlob(ctx, historyIterator)
 		if err != nil {
-			if common.IsNotFoundError(err) {
+			if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
 				// workflow history no longer exists, may due to duplicated archival signal
 				// this may happen even in the middle of iterating history as two archival signals
 				// can be processed concurrently.

common/archiver/filestore/historyArchiver_test.go

@@ -33,6 +33,7 @@ import (
 	"time"
 
 	enumspb "go.temporal.io/api/enums/v1"
+
 	"go.temporal.io/server/tests/testhelper"
 
 	"github.com/golang/mock/gomock"
@@ -294,7 +295,7 @@ func (s *historyArchiverSuite) TestArchive_Skip() {
 		historyIterator.EXPECT().HasNext().Return(true),
 		historyIterator.EXPECT().Next().Return(historyBlob, nil),
 		historyIterator.EXPECT().HasNext().Return(true),
-		historyIterator.EXPECT().Next().Return(nil, &serviceerror.NotFound{Message: "workflow not found"}),
+		historyIterator.EXPECT().Next().Return(nil, serviceerror.NewNotFound("workflow not found")),
 	)
 
 	historyArchiver := s.newTestHistoryArchiver(historyIterator)

common/archiver/gcloud/historyArchiver.go

@@ -153,7 +153,7 @@ func (h *historyArchiver) Archive(ctx context.Context, URI archiver.URI, request
 		part := progress.CurrentPageNumber
 		historyBlob, err := getNextHistoryBlob(ctx, historyIterator)
 		if err != nil {
-			if common.IsNotFoundError(err) {
+			if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
 				// workflow history no longer exists, may due to duplicated archival signal
 				// this may happen even in the middle of iterating history as two archival signals
 				// can be processed concurrently.

common/archiver/gcloud/historyArchiver_test.go

@@ -332,7 +332,7 @@ func (h *historyArchiverSuite) TestArchive_Skip() {
 		historyIterator.EXPECT().HasNext().Return(true),
 		historyIterator.EXPECT().Next().Return(historyBlob, nil),
 		historyIterator.EXPECT().HasNext().Return(true),
-		historyIterator.EXPECT().Next().Return(nil, &serviceerror.NotFound{Message: "workflow not found"}),
+		historyIterator.EXPECT().Next().Return(nil, serviceerror.NewNotFound("workflow not found")),
 	)
 
 	historyArchiver := newHistoryArchiver(h.container, historyIterator, storageWrapper)

common/archiver/historyIterator.go

@@ -172,7 +172,7 @@ func (i *historyIterator) readHistoryBatches(firstEventID int64) ([]*historypb.H
 	newIterState := historyIteratorState{}
 	for size < targetSize {
 		currHistoryBatches, err := i.readHistory(firstEventID)
-		if _, ok := err.(*serviceerror.NotFound); ok && firstEventID != common.FirstEventID {
+		if _, isNotFound := err.(*serviceerror.NotFound); isNotFound && firstEventID != common.FirstEventID {
 			newIterState.FinishedIteration = true
 			return historyBatches, newIterState, nil
 		}
@@ -201,7 +201,7 @@ func (i *historyIterator) readHistoryBatches(firstEventID int64) ([]*historypb.H
 	// If you are here, it means the target size is met after adding the last batch of read history.
 	// We need to check if there's more history batches.
 	_, err := i.readHistory(firstEventID)
-	if _, ok := err.(*serviceerror.NotFound); ok && firstEventID != common.FirstEventID {
+	if _, isNotFound := err.(*serviceerror.NotFound); isNotFound && firstEventID != common.FirstEventID {
 		newIterState.FinishedIteration = true
 		return historyBatches, newIterState, nil
 	}

common/archiver/s3store/historyArchiver.go

@@ -166,7 +166,7 @@ func (h *historyArchiver) Archive(
 	for historyIterator.HasNext() {
 		historyBlob, err := getNextHistoryBlob(ctx, historyIterator)
 		if err != nil {
-			if common.IsNotFoundError(err) {
+			if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
 				// workflow history no longer exists, may due to duplicated archival signal
 				// this may happen even in the middle of iterating history as two archival signals
 				// can be processed concurrently.

common/archiver/s3store/historyArchiver_test.go

@@ -425,7 +425,7 @@ func (s *historyArchiverSuite) TestArchive_Skip() {
 		historyIterator.EXPECT().HasNext().Return(true),
 		historyIterator.EXPECT().Next().Return(historyBlob, nil),
 		historyIterator.EXPECT().HasNext().Return(true),
-		historyIterator.EXPECT().Next().Return(nil, &serviceerror.NotFound{Message: "workflow not found"}),
+		historyIterator.EXPECT().Next().Return(nil, serviceerror.NewNotFound("workflow not found")),
 	)
 
 	historyArchiver := s.newTestHistoryArchiver(historyIterator)

common/namespace/handler.go

@@ -146,7 +146,7 @@ func (d *HandlerImpl) RegisterNamespace(
 	case nil:
 		// namespace already exists, cannot proceed
 		return nil, serviceerror.NewNamespaceAlreadyExists("Namespace already exists.")
-	case *serviceerror.NotFound:
+	case *serviceerror.NamespaceNotFound:
 		// namespace does not exists, proceeds
 	default:
 		// other err

common/namespace/handler_GlobalNamespaceEnabled_NotMasterCluster_test.go

@@ -427,7 +427,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledNotMasterClusterSuite) TestRegist
 		Namespace: namespace,
 	})
 	s.Error(err)
-	s.IsType(&serviceerror.NotFound{}, err)
+	s.IsType(&serviceerror.NamespaceNotFound{}, err)
 	s.Nil(resp)
 }
 
@@ -469,7 +469,7 @@ func (s *namespaceHandlerGlobalNamespaceEnabledNotMasterClusterSuite) TestRegist
 		Namespace: namespace,
 	})
 	s.Error(err)
-	s.IsType(&serviceerror.NotFound{}, err)
+	s.IsType(&serviceerror.NamespaceNotFound{}, err)
 	s.Nil(resp)
 }
 

common/namespace/registry.go

@@ -28,13 +28,13 @@ package namespace
 
 import (
 	"context"
-	"fmt"
 	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"go.temporal.io/api/serviceerror"
+
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/cache"
 	"go.temporal.io/server/common/clock"
@@ -95,7 +95,7 @@ type (
 
 		// GetNamespace reads the state for a single namespace by name or ID
 		// from persistent storage, returning an instance of
-		// serviceerror.NotFound if there is no matching Namespace.
+		// serviceerror.NamespaceNotFound if there is no matching Namespace.
 		GetNamespace(
 			context.Context,
 			*persistence.GetNamespaceRequest,
@@ -484,8 +484,7 @@ func (r *registry) getNamespace(name Name) (*Namespace, error) {
 	if id, ok := r.cacheNameToID.Get(name).(ID); ok {
 		return r.getNamespaceByIDLocked(id)
 	}
-	return nil, serviceerror.NewNotFound(
-		fmt.Sprintf("Namespace name %q not found", name))
+	return nil, serviceerror.NewNamespaceNotFound(name.String())
 }
 
 // getNamespaceByID retrieves the information from the cache if it exists.
@@ -499,8 +498,7 @@ func (r *registry) getNamespaceByIDLocked(id ID) (*Namespace, error) {
 	if ns, ok := r.cacheByID.Get(id).(*Namespace); ok {
 		return ns, nil
 	}
-	return nil, serviceerror.NewNotFound(
-		fmt.Sprintf("Namespace id %q not found", id))
+	return nil, serviceerror.NewNamespaceNotFound(id.String())
 }
 
 func (r *registry) publishCacheUpdate(

common/namespace/registry_test.go

@@ -35,6 +35,7 @@ import (
 	enumspb "go.temporal.io/api/enums/v1"
 	namespacepb "go.temporal.io/api/namespace/v1"
 	"go.temporal.io/api/serviceerror"
+
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/dynamicconfig"
@@ -520,7 +521,7 @@ func (s *registrySuite) TestGetTriggerListAndUpdateCache_ConcurrentAccess() {
 		case nil:
 			s.Equal(entryOld, entryNew)
 			waitGroup.Done()
-		case *serviceerror.NotFound:
+		case *serviceerror.NamespaceNotFound:
 			time.Sleep(4 * time.Second)
 			entryNew, err := s.registry.GetNamespaceByID(id)
 			s.NoError(err)
@@ -637,7 +638,7 @@ func (s *registrySuite) TestRemoveDeletedNamespace() {
 	ns1FromRegistry, err := s.registry.GetNamespace(namespace.Name(namespaceRecord1.Namespace.Info.Name))
 	s.Nil(ns1FromRegistry)
 	s.Error(err)
-	var notFound *serviceerror.NotFound
+	var notFound *serviceerror.NamespaceNotFound
 	s.ErrorAs(err, &notFound)
 }
 

common/namespace/replicationTaskExecutor.go

@@ -161,7 +161,7 @@ func (h *namespaceReplicationTaskExecutorImpl) handleNamespaceCreationReplicatio
 			if resp.Namespace.Info.Id != task.GetId() {
 				return ErrNameUUIDCollision
 			}
-		case *serviceerror.NotFound:
+		case *serviceerror.NamespaceNotFound:
 			// no check is necessary
 			recordExists = false
 		default:
@@ -177,7 +177,7 @@ func (h *namespaceReplicationTaskExecutorImpl) handleNamespaceCreationReplicatio
 			if resp.Namespace.Info.Name != task.Info.GetName() {
 				return ErrNameUUIDCollision
 			}
-		case *serviceerror.NotFound:
+		case *serviceerror.NamespaceNotFound:
 			// no check is necessary
 			recordExists = false
 		default:
@@ -219,7 +219,7 @@ func (h *namespaceReplicationTaskExecutorImpl) handleNamespaceUpdateReplicationT
 		Name: task.Info.GetName(),
 	})
 	if err != nil {
-		if _, ok := err.(*serviceerror.NotFound); ok {
+		if _, isNotFound := err.(*serviceerror.NamespaceNotFound); isNotFound {
 			// this can happen if the create namespace replication task is to processed.
 			// e.g. new cluster which does not have anything
 			return h.handleNamespaceCreationReplicationTask(ctx, task)

common/persistence/cassandra/metadata_store.go

@@ -288,7 +288,7 @@ func (m *MetadataStore) GetNamespace(
 			if len(ID) > 0 {
 				identity = ID
 			}
-			return serviceerror.NewNotFound(fmt.Sprintf("Namespace %s does not exist.", identity))
+			return serviceerror.NewNamespaceNotFound(identity)
 		}
 		return serviceerror.NewUnavailable(fmt.Sprintf("GetNamespace operation failed. Error %v", err))
 	}

common/persistence/cassandra/mutable_state_store.go

@@ -879,7 +879,7 @@ func (d *MutableStateStore) assertNotCurrentExecution(
 		NamespaceID: namespaceID,
 		WorkflowID:  workflowID,
 	}); err != nil {
-		if _, ok := err.(*serviceerror.NotFound); ok {
+		if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
 			// allow bypassing no current record
 			return nil
 		}

common/persistence/clusterMetadataStore.go

@@ -30,6 +30,7 @@ import (
 
 	enumspb "go.temporal.io/api/enums/v1"
 	"go.temporal.io/api/serviceerror"
+
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/persistence/serialization"
@@ -59,7 +60,7 @@ type (
 
 var _ ClusterMetadataManager = (*clusterMetadataManagerImpl)(nil)
 
-//NewClusterMetadataManagerImpl returns new ClusterMetadataManager
+// NewClusterMetadataManagerImpl returns new ClusterMetadataManager
 func NewClusterMetadataManagerImpl(
 	persistence ClusterMetadataStore,
 	serializer serialization.Serializer,
@@ -183,7 +184,7 @@ func (m *clusterMetadataManagerImpl) SaveClusterMetadata(
 	}
 
 	oldClusterMetadata, err := m.GetClusterMetadata(ctx, &GetClusterMetadataRequest{ClusterName: request.GetClusterName()})
-	if _, notFound := err.(*serviceerror.NotFound); notFound {
+	if _, isNotFound := err.(*serviceerror.NotFound); isNotFound {
 		return m.persistence.SaveClusterMetadata(ctx, &InternalSaveClusterMetadataRequest{
 			ClusterName:     request.ClusterName,
 			ClusterMetadata: mcm,

common/persistence/persistence-tests/executionManagerTest.go

@@ -1574,8 +1574,7 @@ func (s *ExecutionManagerSuite) TestDeleteCurrentWorkflow() {
 	runID0, err1 = s.GetCurrentWorkflowRunID(s.ctx, namespaceID, workflowExecution.GetWorkflowId())
 	s.Error(err1)
 	s.Empty(runID0)
-	_, ok := err1.(*serviceerror.NotFound)
-	s.True(ok)
+	s.IsType(&serviceerror.NotFound{}, err1)
 
 	// execution record should still be there
 	_, err2 = s.GetWorkflowMutableState(s.ctx, namespaceID, workflowExecution)
@@ -1623,12 +1622,11 @@ func (s *ExecutionManagerSuite) TestUpdateDeleteWorkflow() {
 	runID0, err1 = s.GetCurrentWorkflowRunID(s.ctx, namespaceID, workflowExecution.GetWorkflowId())
 	s.Error(err1)
 	s.Empty(runID0)
-	_, ok := err1.(*serviceerror.NotFound)
+	s.IsType(&serviceerror.NotFound{}, err1)
 	// execution record should still be there
 	_, err2 = s.GetWorkflowMutableState(s.ctx, namespaceID, workflowExecution)
 	s.Error(err2)
-	_, ok = err2.(*serviceerror.NotFound)
-	s.True(ok)
+	s.IsType(&serviceerror.NotFound{}, err2)
 }
 
 // TestCleanupCorruptedWorkflow test
@@ -1656,8 +1654,7 @@ func (s *ExecutionManagerSuite) TestCleanupCorruptedWorkflow() {
 	runID0, err4 := s.GetCurrentWorkflowRunID(s.ctx, namespaceID, workflowExecution.GetWorkflowId())
 	s.Error(err4)
 	s.Empty(runID0)
-	_, ok := err4.(*serviceerror.NotFound)
-	s.True(ok)
+	s.IsType(&serviceerror.NotFound{}, err4)
 
 	// we should still be able to load with runID
 	info1, err5 := s.GetWorkflowMutableState(s.ctx, namespaceID, workflowExecution)
@@ -1697,8 +1694,7 @@ func (s *ExecutionManagerSuite) TestCleanupCorruptedWorkflow() {
 	// execution record should be gone
 	_, err9 := s.GetWorkflowMutableState(s.ctx, namespaceID, workflowExecution)
 	s.Error(err9)
-	_, ok = err9.(*serviceerror.NotFound)
-	s.True(ok)
+	s.IsType(&serviceerror.NotFound{}, err9)
 }
 
 // TestGetCurrentWorkflow test

common/persistence/persistence-tests/metadataPersistenceV2Test.go

@@ -215,7 +215,7 @@ func (m *MetadataPersistenceSuiteV2) TestGetNamespace() {
 	resp0, err0 := m.GetNamespace("", "does-not-exist")
 	m.Nil(resp0)
 	m.Error(err0)
-	m.IsType(&serviceerror.NotFound{}, err0)
+	m.IsType(&serviceerror.NamespaceNotFound{}, err0)
 	testBinaries := &namespacepb.BadBinaries{
 		Binaries: map[string]*namespacepb.BadBinaryInfo{
 			"abc": {
@@ -935,12 +935,12 @@ func (m *MetadataPersistenceSuiteV2) TestDeleteNamespace() {
 		time.Sleep(time.Second * time.Duration(i))
 	}
 	m.Error(err4)
-	m.IsType(&serviceerror.NotFound{}, err4)
+	m.IsType(&serviceerror.NamespaceNotFound{}, err4)
 	m.Nil(resp4)
 
 	resp5, err5 := m.GetNamespace(id, "")
 	m.Error(err5)
-	m.IsType(&serviceerror.NotFound{}, err5)
+	m.IsType(&serviceerror.NamespaceNotFound{}, err5)
 	m.Nil(resp5)
 
 	id = uuid.New()
@@ -976,12 +976,12 @@ func (m *MetadataPersistenceSuiteV2) TestDeleteNamespace() {
 
 	resp8, err8 := m.GetNamespace("", name)
 	m.Error(err8)
-	m.IsType(&serviceerror.NotFound{}, err8)
+	m.IsType(&serviceerror.NamespaceNotFound{}, err8)
 	m.Nil(resp8)
 
 	resp9, err9 := m.GetNamespace(id, "")
 	m.Error(err9)
-	m.IsType(&serviceerror.NotFound{}, err9)
+	m.IsType(&serviceerror.NamespaceNotFound{}, err9)
 	m.Nil(resp9)
 }
 

common/persistence/persistenceMetricClients.go

@@ -1399,7 +1399,7 @@ func (p *metricEmitter) updateErrorMetric(scope int, err error) {
 		p.metricClient.IncCounter(scope, metrics.PersistenceErrBadRequestCounter)
 	case *serviceerror.NamespaceAlreadyExists:
 		p.metricClient.IncCounter(scope, metrics.PersistenceErrNamespaceAlreadyExistsCounter)
-	case *serviceerror.NotFound:
+	case *serviceerror.NotFound, *serviceerror.NamespaceNotFound:
 		p.metricClient.IncCounter(scope, metrics.PersistenceErrEntityNotExistsCounter)
 	case *serviceerror.ResourceExhausted:
 		p.metricClient.IncCounter(scope, metrics.PersistenceErrBusyCounter)

common/persistence/sql/metadata.go

@@ -124,7 +124,7 @@ func (m *sqlMetadataManagerV2) GetNamespace(
 				identity = request.ID
 			}
 
-			return nil, serviceerror.NewNotFound(fmt.Sprintf("Namespace %s does not exist.", identity))
+			return nil, serviceerror.NewNamespaceNotFound(identity)
 		default:
 			return nil, serviceerror.NewUnavailable(fmt.Sprintf("GetNamespace operation failed. Error %v", err))
 		}