diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go
index 876ec9c..9e6ed94 100644
--- a/apigeeSync_suite_test.go
+++ b/apigeeSync_suite_test.go
@@ -1,7 +1,6 @@
 package apidApigeeSync
 
 import (
-	"github.com/apigee-labs/transicator/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
@@ -12,17 +11,25 @@
 	"time"
 
 	"github.com/30x/apid-core"
+
 	"github.com/30x/apid-core/factory"
 )
 
 var (
-	tmpDir     string
-	testServer *httptest.Server
-	testRouter apid.Router
-	testMock   *MockServer
+	tmpDir         string
+	testServer     *httptest.Server
+	testRouter     apid.Router
+	testMock       *MockServer
+	wipeDBAferTest bool
 )
 
-var _ = BeforeSuite(func(done Done) {
+const dummyConfigValue string = "placeholder"
+
+var _ = BeforeSuite(func() {
+	wipeDBAferTest = true
+})
+
+var _ = BeforeEach(func(done Done) {
 	apid.Initialize(factory.DefaultServicesFactory())
 
 	config = apid.Config()
@@ -32,12 +39,9 @@
 	Expect(err).NotTo(HaveOccurred())
 	config.Set("local_storage_path", tmpDir)
 
-	testRouter = apid.API().Router()
-	testServer = httptest.NewServer(testRouter)
-
-	config.Set(configProxyServerBaseURI, testServer.URL)
-	config.Set(configSnapServerBaseURI, testServer.URL)
-	config.Set(configChangeServerBaseURI, testServer.URL)
+	config.Set(configProxyServerBaseURI, dummyConfigValue)
+	config.Set(configSnapServerBaseURI, dummyConfigValue)
+	config.Set(configChangeServerBaseURI, dummyConfigValue)
 	config.Set(configSnapshotProtocol, "json")
 	config.Set(configPollInterval, 10*time.Millisecond)
 
@@ -49,140 +53,27 @@
 	block = "0"
 	log = apid.Log()
 
-	// set up mock server
-	mockParms := MockParms{
-		ReliableAPI:  false,
-		ClusterID:    config.GetString(configApidClusterId),
-		TokenKey:     config.GetString(configConsumerKey),
-		TokenSecret:  config.GetString(configConsumerSecret),
-		Scope:        "ert452",
-		Organization: "att",
-		Environment:  "prod",
-	}
-	testMock = Mock(mockParms, testRouter)
+	_initPlugin(apid.AllServices())
+	close(done)
+}, 3)
 
-	// This is actually the first test :)
-	// Tests that entire bootstrap and set of sync operations work
-	var lastSnapshot *common.Snapshot
-
-	expectedSnapshotTables := make(map[string]bool)
-	expectedSnapshotTables["kms.company"] = true
-	expectedSnapshotTables["edgex.apid_cluster"] = true
-	expectedSnapshotTables["edgex.data_scope"] = true
-
-	apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-		defer GinkgoRecover()
-
-		if s, ok := event.(*common.Snapshot); ok {
-
-			//verify that during downloadDataSnapshot, knownTables was correctly populated
-			Expect(mapIsSubset(knownTables, expectedSnapshotTables)).To(BeTrue())
-
-			/* After this, we will mock changes for tables not present in the initial snapshot
-			* until that is changed in the mock server, we have to spoof the known tables
-			 */
-
-			//add apid_cluster and data_scope since those would present if this were a real scenario
-			knownTables["kms.app_credential"] = true
-			knownTables["kms.app_credential_apiproduct_mapper"] = true
-			knownTables["kms.developer"] = true
-			knownTables["kms.company_developer"] = true
-			knownTables["kms.api_product"] = true
-			knownTables["kms.app"] = true
-
-			lastSnapshot = s
-
-			for _, t := range s.Tables {
-				switch t.Name {
-
-				case "edgex.apid_cluster":
-					Expect(t.Rows).To(HaveLen(1))
-					r := t.Rows[0]
-					var id string
-					r.Get("id", &id)
-					Expect(id).To(Equal("bootstrap"))
-
-				case "edgex.data_scope":
-					Expect(t.Rows).To(HaveLen(2))
-					r := t.Rows[1] // get the non-cluster row
-
-					var id, clusterID, env, org, scope string
-					r.Get("id", &id)
-					r.Get("apid_cluster_id", &clusterID)
-					r.Get("env", &env)
-					r.Get("org", &org)
-					r.Get("scope", &scope)
-
-					Expect(id).To(Equal("ert452"))
-					Expect(scope).To(Equal("ert452"))
-					Expect(clusterID).To(Equal("bootstrap"))
-					Expect(env).To(Equal("prod"))
-					Expect(org).To(Equal("att"))
-				}
-			}
-
-		} else if cl, ok := event.(*common.ChangeList); ok {
-
-			// ensure that snapshot switched DB versions
-			Expect(apidInfo.LastSnapshot).To(Equal(lastSnapshot.SnapshotInfo))
-			expectedDB, err := data.DBVersion(lastSnapshot.SnapshotInfo)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(getDB() == expectedDB).Should(BeTrue())
-
-			Expect(cl.Changes).To(HaveLen(6))
-
-			var tables []string
-			for _, c := range cl.Changes {
-				tables = append(tables, c.Table)
-				Expect(c.NewRow).ToNot(BeNil())
-
-				var tenantID string
-				c.NewRow.Get("tenant_id", &tenantID)
-				Expect(tenantID).To(Equal("ert452"))
-			}
-
-			Expect(tables).To(ContainElement("kms.app_credential"))
-			Expect(tables).To(ContainElement("kms.app_credential_apiproduct_mapper"))
-			Expect(tables).To(ContainElement("kms.developer"))
-			Expect(tables).To(ContainElement("kms.company_developer"))
-			Expect(tables).To(ContainElement("kms.api_product"))
-			Expect(tables).To(ContainElement("kms.app"))
-
-			events.ListenFunc(apid.EventDeliveredSelector, func(e apid.Event) {
-				defer GinkgoRecover()
-
-				// allow other handler to execute to insert last_sequence
-				time.Sleep(50 * time.Millisecond)
-				var seq string
-				err = getDB().
-					QueryRow("SELECT last_sequence FROM APID_CLUSTER LIMIT 1;").
-					Scan(&seq)
-
-				Expect(err).NotTo(HaveOccurred())
-				Expect(seq).To(Equal(cl.LastSequence))
-
-				close(done)
-			})
-		}
-	})
-
-	apid.InitializePlugins()
-})
-
-var _ = BeforeEach(func() {
+var _ = AfterEach(func() {
 	apid.Events().Close()
 
 	lastSequence = ""
 
-	_, err := getDB().Exec("DELETE FROM APID_CLUSTER")
-	Expect(err).NotTo(HaveOccurred())
-	_, err = getDB().Exec("DELETE FROM DATA_SCOPE")
-	Expect(err).NotTo(HaveOccurred())
+	if wipeDBAferTest {
+		_, err := getDB().Exec("DELETE FROM APID_CLUSTER")
+		Expect(err).NotTo(HaveOccurred())
+		_, err = getDB().Exec("DELETE FROM DATA_SCOPE")
+		Expect(err).NotTo(HaveOccurred())
 
-	db, err := data.DB()
-	Expect(err).NotTo(HaveOccurred())
-	_, err = db.Exec("DELETE FROM APID")
-	Expect(err).NotTo(HaveOccurred())
+		db, err := dataService.DB()
+		Expect(err).NotTo(HaveOccurred())
+		_, err = db.Exec("DELETE FROM APID")
+		Expect(err).NotTo(HaveOccurred())
+	}
+	wipeDBAferTest = true
 })
 
 var _ = AfterSuite(func() {
diff --git a/apigee_sync.go b/apigee_sync.go
index be3fc8c..d6b3a19 100644
--- a/apigee_sync.go
+++ b/apigee_sync.go
@@ -1,514 +1,99 @@
 package apidApigeeSync
 
 import (
-	"encoding/json"
-	"net/http"
-	"net/url"
-	"path"
-	"time"
-
-	"sync/atomic"
-
-	"io/ioutil"
-
 	"github.com/30x/apid-core"
-	"github.com/apigee-labs/transicator/common"
+	"net/http"
+	"time"
 )
 
 const (
-	httpTimeout       = time.Minute
-	pluginTimeout     = time.Minute
-	maxBackoffTimeout = time.Minute
+	httpTimeout   = time.Minute
+	pluginTimeout = time.Minute
 )
 
-var (
-	block        string = "45"
-	lastSequence string
-	polling      uint32
-	knownTables  = make(map[string]bool)
-)
+var knownTables = make(map[string]bool)
 
 /*
- * Polls change agent for changes. In event of errors, uses a doubling
- * backoff from 200ms up to a max delay of the configPollInterval value.
+ *  Start from existing snapshot if possible
+ *  If an existing snapshot does not exist, use the apid scope to fetch
+ *  all data scopes, then get a snapshot for those data scopes
+ *
+ *  Then, poll for changes
  */
-func pollForChanges() {
+func bootstrap() {
 
-	if atomic.SwapUint32(&polling, 1) == 1 {
+	if apidInfo.LastSnapshot != "" {
+		snapshot := startOnLocalSnapshot(apidInfo.LastSnapshot)
+
+		events.EmitWithCallback(ApigeeSyncEventSelector, snapshot, func(event apid.Event) {
+			changeManager.pollChangeWithBackoff()
+		})
+
+		log.Infof("Started on local snapshot: %s", snapshot.SnapshotInfo)
 		return
 	}
 
-	var backOffFunc func()
-	pollInterval := config.GetDuration(configPollInterval)
-	for {
-		start := time.Now()
-		err := pollChangeAgent()
-		end := time.Now()
-		if err != nil {
-			if _, ok := err.(changeServerError); ok {
-				downloadDataSnapshot()
-				continue
-			}
-			log.Debugf("Error connecting to changeserver: %v", err)
-		}
-		if end.After(start.Add(time.Second)) {
-			backOffFunc = nil
-			continue
-		}
-		if backOffFunc == nil {
-			backOffFunc = createBackOff(200*time.Millisecond, pollInterval)
-		}
-		backOffFunc()
-	}
+	downloadBootSnapshot(nil)
+	downloadDataSnapshot(quitPollingSnapshotServer)
 
-	atomic.SwapUint32(&polling, 0)
+	changeManager.pollChangeWithBackoff()
+
 }
 
 /*
- * Long polls the change agent with a 45 second block. Parses the response from
- * change agent and raises an event. Called by pollForChanges().
+ * Call toExecute repeatedly until it does not return an error, with an exponential backoff policy
+ * for retrying on errors
  */
-func pollChangeAgent() error {
+func pollWithBackoff(quit chan bool, toExecute func(chan bool) error, handleError func(error)) {
 
-	changesUri, err := url.Parse(config.GetString(configChangeServerBaseURI))
-	if err != nil {
-		log.Errorf("bad url value for config %s: %s", changesUri, err)
-		return err
-	}
-	changesUri.Path = path.Join(changesUri.Path, "changes")
+	backoff := NewExponentialBackoff(200*time.Millisecond, config.GetDuration(configPollInterval), 2, true)
 
-	/*
-	 * Check to see if we have lastSequence already saved in the DB,
-	 * in which case, it has to be used to prevent re-reading same data
-	 */
-	lastSequence = getLastSequence()
+	//inintialize the retry channel to start first attempt immediately
+	retry := time.After(0 * time.Millisecond)
+
 	for {
-		log.Debug("polling...")
+		select {
+		case <-quit:
+			log.Info("Quit signal recieved.  Returning")
+			return
+		case <-retry:
+			start := time.Now()
 
-		/* Find the scopes associated with the config id */
-		scopes := scopeCache.readAllScope()
-		v := url.Values{}
-
-		/* Sequence added to the query if available */
-		if lastSequence != "" {
-			v.Add("since", lastSequence)
-		}
-		v.Add("block", block)
-
-		/*
-		 * Include all the scopes associated with the config Id
-		 * The Config Id is included as well, as it acts as the
-		 * Bootstrap scope
-		 */
-		for _, scope := range scopes {
-			v.Add("scope", scope)
-		}
-		v.Add("scope", apidInfo.ClusterID)
-		v.Add("snapshot", apidInfo.LastSnapshot)
-		changesUri.RawQuery = v.Encode()
-		uri := changesUri.String()
-		log.Debugf("Fetching changes: %s", uri)
-
-		/* If error, break the loop, and retry after interval */
-		client := &http.Client{Timeout: httpTimeout} // must be greater than block value
-		req, err := http.NewRequest("GET", uri, nil)
-		addHeaders(req)
-
-		r, err := client.Do(req)
-		if err != nil {
-			log.Errorf("change agent comm error: %s", err)
-			return err
-		}
-
-		if r.StatusCode != http.StatusOK {
-			log.Errorf("Get changes request failed with status code: %d", r.StatusCode)
-			switch r.StatusCode {
-			case http.StatusUnauthorized:
-				tokenManager.invalidateToken()
-
-			case http.StatusNotModified:
-				r.Body.Close()
-				continue
-
-			case http.StatusBadRequest:
-				var apiErr changeServerError
-				var b []byte
-				b, err = ioutil.ReadAll(r.Body)
-				if err != nil {
-					log.Errorf("Unable to read response body: %v", err)
-					break
-				}
-				err = json.Unmarshal(b, &apiErr)
-				if err != nil {
-					log.Errorf("JSON Response Data not parsable: %s", string(b))
-					break
-				}
-				if apiErr.Code == "SNAPSHOT_TOO_OLD" {
-					log.Debug("Received SNAPSHOT_TOO_OLD message from change server.")
-					err = apiErr
-				}
+			err := toExecute(quit)
+			if err == nil {
+				return
 			}
 
-			r.Body.Close()
-			return err
-		}
+			if _, ok := err.(quitSignalError); ok {
+				return
+			}
 
-		var resp common.ChangeList
-		err = json.NewDecoder(r.Body).Decode(&resp)
-		r.Body.Close()
-		if err != nil {
-			log.Errorf("JSON Response Data not parsable: %v", err)
-			return err
-		}
+			end := time.Now()
+			//error encountered, since we would have returned above otherwise
+			handleError(err)
 
-		/*
-		 * If the lastSequence is already newer or the same than what we got via
-		 * resp.LastSequence, Ignore it.
-		 */
-		if lastSequence != "" &&
-			getChangeStatus(lastSequence, resp.LastSequence) != 1 {
-			log.Errorf("Ignore change, already have newer changes")
-			continue
-		}
-
-		if changesRequireDDLSync(resp) {
-			log.Info("Detected DDL changes, going to fetch a new snapshot to sync...")
-			return changeServerError{
-				Code: "DDL changes detected; must get new snapshot",
+			/* TODO keep this around? Imagine an immediately erroring service,
+			 *  causing many sequential requests which could pollute logs
+			 */
+			//only backoff if the request took less than one second
+			if end.After(start.Add(time.Second)) {
+				backoff.Reset()
+				retry = time.After(0 * time.Millisecond)
+			} else {
+				retry = time.After(backoff.Duration())
 			}
 		}
-
-		/* If valid data present, Emit to plugins */
-		if len(resp.Changes) > 0 {
-			done := make(chan bool)
-			events.EmitWithCallback(ApigeeSyncEventSelector, &resp, func(event apid.Event) {
-				done <- true
-			})
-
-			select {
-			case <-time.After(httpTimeout):
-				log.Panic("Timeout. Plugins failed to respond to changes.")
-			case <-done:
-			}
-		} else {
-			log.Debugf("No Changes detected for Scopes: %s", scopes)
-		}
-
-		updateSequence(resp.LastSequence)
 	}
 }
 
-/*
- * seqCurr.Compare() will return 1, if its newer than seqPrev,
- * else will return 0, if same, or -1 if older.
- */
-func getChangeStatus(lastSeq string, currSeq string) int {
-	seqPrev, err := common.ParseSequence(lastSeq)
-	if err != nil {
-		log.Panic("Unable to parse previous sequence string")
-	}
-	seqCurr, err := common.ParseSequence(currSeq)
-	if err != nil {
-		log.Panic("Unable to parse current sequence string")
-	}
-	return seqCurr.Compare(seqPrev)
-}
-
-func updateSequence(seq string) {
-	lastSequence = seq
-	err := updateLastSequence(seq)
-	if err != nil {
-		log.Panic("Unable to update Sequence in DB")
-	}
-
-}
-
-func changesRequireDDLSync(changes common.ChangeList) bool {
-
-	return !mapIsSubset(knownTables, extractTablesFromChangelist(changes))
-}
-
-// simple doubling back-off
-func createBackOff(retryIn, maxBackOff time.Duration) func() {
-	return func() {
-		if retryIn > maxBackOff {
-			retryIn = maxBackOff
-		}
-		log.Debugf("backoff called. will retry in %s.", retryIn)
-		time.Sleep(retryIn)
-		retryIn = retryIn * time.Duration(2)
-	}
-}
-
-func Redirect(req *http.Request, via []*http.Request) error {
+func Redirect(req *http.Request, _ []*http.Request) error {
 	req.Header.Add("Authorization", "Bearer "+tokenManager.getBearerToken())
 	req.Header.Add("org", apidInfo.ClusterID) // todo: this is strange.. is it needed?
 	return nil
 }
 
-// pollForChanges should usually be true, tests use the flag
-func bootstrap() {
-
-	if apidInfo.LastSnapshot != "" {
-		startOnLocalSnapshot(apidInfo.LastSnapshot)
-		return
-	}
-
-	downloadBootSnapshot()
-	downloadDataSnapshot()
-	go pollForChanges()
-}
-
-// retrieve boot information: apid_config and apid_config_scope
-func downloadBootSnapshot() {
-	log.Debug("download Snapshot for boot data")
-
-	scopes := []string{apidInfo.ClusterID}
-	snapshot := downloadSnapshot(scopes)
-	storeBootSnapshot(snapshot)
-}
-
-func storeBootSnapshot(snapshot common.Snapshot) {
-	// note that for boot snapshot case, we don't touch databases. We only update in-mem cache
-	// This aims to deal with duplicate snapshot version#, see XAPID-869 for details
-	scopeCache.clearAndInitCache(snapshot.SnapshotInfo)
-	for _, table := range snapshot.Tables {
-		if table.Name == LISTENER_TABLE_DATA_SCOPE {
-			for _, row := range table.Rows {
-				ds := makeDataScopeFromRow(row)
-				// cache scopes for this cluster
-				if ds.ClusterID == apidInfo.ClusterID {
-					scopeCache.updateCache(&ds)
-				}
-			}
-		}
-	}
-	// note that for boot snapshot case, we don't need to inform plugins as they'll get the data snapshot
-}
-
-// use the scope IDs from the boot snapshot to get all the data associated with the scopes
-func downloadDataSnapshot() {
-	log.Debug("download Snapshot for data scopes")
-
-	scopes := scopeCache.readAllScope()
-
-	scopes = append(scopes, apidInfo.ClusterID)
-	snapshot := downloadSnapshot(scopes)
-	storeDataSnapshot(snapshot)
-}
-
-func storeDataSnapshot(snapshot common.Snapshot) {
-	knownTables = extractTablesFromSnapshot(snapshot)
-
-	db, err := data.DBVersion(snapshot.SnapshotInfo)
-	if err != nil {
-		log.Panicf("Database inaccessible: %v", err)
-	}
-	persistKnownTablesToDB(knownTables, db)
-
-	done := make(chan bool)
-	log.Info("Emitting Snapshot to plugins")
-	events.EmitWithCallback(ApigeeSyncEventSelector, &snapshot, func(event apid.Event) {
-		done <- true
-	})
-
-	select {
-	case <-time.After(pluginTimeout):
-		log.Panic("Timeout. Plugins failed to respond to snapshot.")
-	case <-done:
-	}
-
-}
-
-func extractTablesFromSnapshot(snapshot common.Snapshot) (tables map[string]bool) {
-
-	tables = make(map[string]bool)
-
-	log.Debug("Extracting table names from snapshot")
-	if snapshot.Tables == nil {
-		//if this panic ever fires, it's a bug
-		log.Panicf("Attempt to extract known tables from snapshot without tables failed")
-	}
-
-	for _, table := range snapshot.Tables {
-		tables[table.Name] = true
-	}
-
-	return tables
-}
-
-func extractTablesFromChangelist(changes common.ChangeList) (tables map[string]bool) {
-
-	tables = make(map[string]bool)
-
-	for _, change := range changes.Changes {
-		tables[change.Table] = true
-	}
-
-	return tables
-}
-
-func extractTablesFromDB(db apid.DB) (tables map[string]bool) {
-
-	tables = make(map[string]bool)
-
-	log.Debug("Extracting table names from existing DB")
-	rows, err := db.Query("SELECT name FROM _known_tables;")
-	defer rows.Close()
-
-	if err != nil {
-		log.Panicf("Error reading current set of tables: %v", err)
-	}
-
-	for rows.Next() {
-		var table string
-		if err := rows.Scan(&table); err != nil {
-			log.Panicf("Error reading current set of tables: %v", err)
-		}
-		log.Debugf("Table %s found in existing db", table)
-
-		tables[table] = true
-	}
-	return tables
-}
-
-func persistKnownTablesToDB(tables map[string]bool, db apid.DB) {
-	log.Debugf("Inserting table names found in snapshot into db")
-
-	tx, err := db.Begin()
-	if err != nil {
-		log.Panicf("Error starting transaction: %v", err)
-	}
-	defer tx.Rollback()
-
-	_, err = tx.Exec("CREATE TABLE _known_tables (name text, PRIMARY KEY(name));")
-	if err != nil {
-		log.Panicf("Could not create _known_tables table: %s", err)
-	}
-
-	for name := range tables {
-		log.Debugf("Inserting %s into _known_tables", name)
-		_, err := tx.Exec("INSERT INTO _known_tables VALUES(?);", name)
-		if err != nil {
-			log.Panicf("Error encountered inserting into known tables ", err)
-		}
-
-	}
-
-	err = tx.Commit()
-	if err != nil {
-		log.Panicf("Error committing transaction: %v", err)
-
-	}
-}
-
-// Skip Downloading snapshot if there is already a snapshot available from previous run
-func startOnLocalSnapshot(snapshot string) {
-	log.Infof("Starting on local snapshot: %s", snapshot)
-
-	// ensure DB version will be accessible on behalf of dependant plugins
-	db, err := data.DBVersion(snapshot)
-	if err != nil {
-		log.Panicf("Database inaccessible: %v", err)
-	}
-
-	knownTables = extractTablesFromDB(db)
-	scopeCache.clearAndInitCache(snapshot)
-
-	// allow plugins (including this one) to start immediately on existing database
-	// Note: this MUST have no tables as that is used as an indicator
-	snap := &common.Snapshot{
-		SnapshotInfo: snapshot,
-	}
-	events.EmitWithCallback(ApigeeSyncEventSelector, snap, func(event apid.Event) {
-		go pollForChanges()
-	})
-
-	log.Infof("Started on local snapshot: %s", snapshot)
-}
-
-// will keep retrying with backoff until success
-func downloadSnapshot(scopes []string) common.Snapshot {
-
-	log.Debug("downloadSnapshot")
-
-	snapshotUri, err := url.Parse(config.GetString(configSnapServerBaseURI))
-	if err != nil {
-		log.Panicf("bad url value for config %s: %s", snapshotUri, err)
-	}
-
-	/* Frame and send the snapshot request */
-	snapshotUri.Path = path.Join(snapshotUri.Path, "snapshots")
-
-	v := url.Values{}
-	for _, scope := range scopes {
-		v.Add("scope", scope)
-	}
-	snapshotUri.RawQuery = v.Encode()
-	uri := snapshotUri.String()
-	log.Infof("Snapshot Download: %s", uri)
-
-	client := &http.Client{
-		CheckRedirect: Redirect,
-		Timeout:       httpTimeout,
-	}
-
-	retryIn := 5 * time.Millisecond
-	maxBackOff := maxBackoffTimeout
-	backOffFunc := createBackOff(retryIn, maxBackOff)
-	first := true
-
-	for {
-		if first {
-			first = false
-		} else {
-			backOffFunc()
-		}
-
-		req, err := http.NewRequest("GET", uri, nil)
-		if err != nil {
-			// should never happen, but if it does, it's unrecoverable anyway
-			log.Panicf("Snapshotserver comm error: %v", err)
-		}
-		addHeaders(req)
-
-		// Set the transport protocol type based on conf file input
-		if config.GetString(configSnapshotProtocol) == "json" {
-			req.Header.Set("Accept", "application/json")
-		} else {
-			req.Header.Set("Accept", "application/proto")
-		}
-
-		// Issue the request to the snapshot server
-		r, err := client.Do(req)
-		if err != nil {
-			log.Errorf("Snapshotserver comm error: %v", err)
-			continue
-		}
-
-		if r.StatusCode != 200 {
-			body, _ := ioutil.ReadAll(r.Body)
-			log.Errorf("Snapshot server conn failed with resp code %d, body: %s", r.StatusCode, string(body))
-			r.Body.Close()
-			continue
-		}
-
-		// Decode the Snapshot server response
-		var resp common.Snapshot
-		err = json.NewDecoder(r.Body).Decode(&resp)
-		if err != nil {
-			log.Errorf("JSON Response Data not parsable: %v", err)
-			r.Body.Close()
-			continue
-		}
-
-		r.Body.Close()
-		return resp
-	}
-}
-
 func addHeaders(req *http.Request) {
-	req.Header.Add("Authorization", "Bearer "+tokenManager.getBearerToken())
+	req.Header.Add("Authorization", "Bearer "+ tokenManager.getBearerToken())
 	req.Header.Set("apid_instance_id", apidInfo.InstanceID)
 	req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
 	req.Header.Set("updated_at_apid", time.Now().Format(time.RFC3339))
@@ -518,25 +103,20 @@
 	Code string `json:"code"`
 }
 
+type quitSignalError struct {
+}
+
+type expected200Error struct {
+}
+
+func (an expected200Error) Error() string {
+	return "Did not recieve OK response"
+}
+
+func (a quitSignalError) Error() string {
+	return "Signal to quit encountered"
+}
+
 func (a changeServerError) Error() string {
 	return a.Code
-}
-
-/*
- * Determine is map b is a subset of map a
- */
-func mapIsSubset(a map[string]bool, b map[string]bool) bool {
-
-	//nil maps should not be passed in.  Making the distinction between nil map and empty map
-	if a == nil || b == nil {
-		return false
-	}
-
-	for k := range b {
-		if !a[k] {
-			return false
-		}
-	}
-
-	return true
-}
+}
\ No newline at end of file
diff --git a/apigee_sync_test.go b/apigee_sync_test.go
index e9af4df..9cdec11 100644
--- a/apigee_sync_test.go
+++ b/apigee_sync_test.go
@@ -5,90 +5,280 @@
 	"github.com/apigee-labs/transicator/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"net/http/httptest"
+	//"time"
 )
 
-var _ = Describe("listener", func() {
+var _ = Describe("Sync", func() {
 
-	It("should bootstrap from local DB if present", func(done Done) {
+	Context("Sync", func() {
 
-		expectedTables := make(map[string]bool)
-		expectedTables["kms.company"] = true
-		expectedTables["edgex.apid_cluster"] = true
-		expectedTables["edgex.data_scope"] = true
+		var initializeContext = func() {
+			testRouter = apid.API().Router()
+			testServer = httptest.NewServer(testRouter)
 
-		Expect(apidInfo.LastSnapshot).NotTo(BeEmpty())
-
-		apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-			defer GinkgoRecover()
-
-			if s, ok := event.(*common.Snapshot); ok {
-
-				//verify that the knownTables array has been properly populated from existing DB
-				Expect(mapIsSubset(knownTables, expectedTables)).To(BeTrue())
-
-				Expect(s.SnapshotInfo).Should(Equal(apidInfo.LastSnapshot))
-				Expect(s.Tables).To(BeNil())
-
-				close(done)
+			// set up mock server
+			mockParms := MockParms{
+				ReliableAPI:  false,
+				ClusterID:    config.GetString(configApidClusterId),
+				TokenKey:     config.GetString(configConsumerKey),
+				TokenSecret:  config.GetString(configConsumerSecret),
+				Scope:        "ert452",
+				Organization: "att",
+				Environment:  "prod",
 			}
+			testMock = Mock(mockParms, testRouter)
+
+			config.Set(configProxyServerBaseURI, testServer.URL)
+			config.Set(configSnapServerBaseURI, testServer.URL)
+			config.Set(configChangeServerBaseURI, testServer.URL)
+		}
+
+		var restoreContext = func() {
+
+			testServer.Close()
+
+			config.Set(configProxyServerBaseURI, dummyConfigValue)
+			config.Set(configSnapServerBaseURI, dummyConfigValue)
+			config.Set(configChangeServerBaseURI, dummyConfigValue)
+
+		}
+
+		It("should succesfully bootstrap from clean slate", func(done Done) {
+			log.Info("Starting sync tests...")
+			var closeDone <-chan bool
+			initializeContext()
+			// do not wipe DB after.  Lets use it
+			wipeDBAferTest = false
+			var lastSnapshot *common.Snapshot
+
+			expectedSnapshotTables := common.ChangeList{
+				Changes: []common.Change{common.Change{Table: "kms.company"},
+					common.Change{Table: "edgex.apid_cluster"},
+					common.Change{Table: "edgex.data_scope"}},
+			}
+
+			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+				if s, ok := event.(*common.Snapshot); ok {
+
+					Expect(changesRequireDDLSync(expectedSnapshotTables)).To(BeFalse())
+
+					//add apid_cluster and data_scope since those would present if this were a real scenario
+					knownTables["kms.app_credential"] = true
+					knownTables["kms.app_credential_apiproduct_mapper"] = true
+					knownTables["kms.developer"] = true
+					knownTables["kms.company_developer"] = true
+					knownTables["kms.api_product"] = true
+					knownTables["kms.app"] = true
+
+					lastSnapshot = s
+
+					for _, t := range s.Tables {
+						switch t.Name {
+
+						case "edgex.apid_cluster":
+							Expect(t.Rows).To(HaveLen(1))
+							r := t.Rows[0]
+							var id string
+							r.Get("id", &id)
+							Expect(id).To(Equal("bootstrap"))
+
+						case "edgex.data_scope":
+							Expect(t.Rows).To(HaveLen(2))
+							r := t.Rows[1] // get the non-cluster row
+
+							var id, clusterID, env, org, scope string
+							r.Get("id", &id)
+							r.Get("apid_cluster_id", &clusterID)
+							r.Get("env", &env)
+							r.Get("org", &org)
+							r.Get("scope", &scope)
+
+							Expect(id).To(Equal("ert452"))
+							Expect(scope).To(Equal("ert452"))
+							Expect(clusterID).To(Equal("bootstrap"))
+							Expect(env).To(Equal("prod"))
+							Expect(org).To(Equal("att"))
+						}
+					}
+
+				} else if cl, ok := event.(*common.ChangeList); ok {
+					closeDone = changeManager.close()
+					// ensure that snapshot switched DB versions
+					Expect(apidInfo.LastSnapshot).To(Equal(lastSnapshot.SnapshotInfo))
+					expectedDB, err := dataService.DBVersion(lastSnapshot.SnapshotInfo)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(getDB() == expectedDB).Should(BeTrue())
+
+					Expect(cl.Changes).To(HaveLen(6))
+
+					var tables []string
+					for _, c := range cl.Changes {
+						tables = append(tables, c.Table)
+						Expect(c.NewRow).ToNot(BeNil())
+
+						var tenantID string
+						c.NewRow.Get("tenant_id", &tenantID)
+						Expect(tenantID).To(Equal("ert452"))
+					}
+
+					Expect(tables).To(ContainElement("kms.app_credential"))
+					Expect(tables).To(ContainElement("kms.app_credential_apiproduct_mapper"))
+					Expect(tables).To(ContainElement("kms.developer"))
+					Expect(tables).To(ContainElement("kms.company_developer"))
+					Expect(tables).To(ContainElement("kms.api_product"))
+					Expect(tables).To(ContainElement("kms.app"))
+
+					go func() {
+						// when close done, all handlers for the first changeList have been executed
+						<-closeDone
+						defer GinkgoRecover()
+						// allow other handler to execute to insert last_sequence
+						var seq string
+						//for seq = ""; seq == ""; {
+						//	time.Sleep(50 * time.Millisecond)
+						err := getDB().
+							QueryRow("SELECT last_sequence FROM APID_CLUSTER LIMIT 1;").
+							Scan(&seq)
+						Expect(err).NotTo(HaveOccurred())
+						//}
+						Expect(seq).To(Equal(cl.LastSequence))
+
+						restoreContext()
+						close(done)
+					}()
+
+				}
+			})
+			pie := apid.PluginsInitializedEvent{
+				Description: "plugins initialized",
+			}
+			pie.Plugins = append(pie.Plugins, pluginData)
+			postInitPlugins(pie)
+		}, 3)
+
+		It("should bootstrap from local DB if present", func(done Done) {
+
+			var closeDone <-chan bool
+
+			initializeContext()
+			expectedTables := common.ChangeList{
+				Changes: []common.Change{common.Change{Table: "kms.company"},
+					common.Change{Table: "edgex.apid_cluster"},
+					common.Change{Table: "edgex.data_scope"}},
+			}
+			Expect(apidInfo.LastSnapshot).NotTo(BeEmpty())
+
+			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+
+				if s, ok := event.(*common.Snapshot); ok {
+					// In this test, the changeManager.pollChangeWithBackoff() has not been launched when changeManager closed
+					// This is because the changeManager.pollChangeWithBackoff() in bootstrap() happened after this handler
+					closeDone = changeManager.close()
+					go func() {
+						// when close done, all handlers for the first snapshot have been executed
+						<-closeDone
+						//verify that the knownTables array has been properly populated from existing DB
+						Expect(changesRequireDDLSync(expectedTables)).To(BeFalse())
+
+						Expect(s.SnapshotInfo).Should(Equal(apidInfo.LastSnapshot))
+						Expect(s.Tables).To(BeNil())
+
+						restoreContext()
+						close(done)
+					}()
+
+				}
+			})
+			pie := apid.PluginsInitializedEvent{
+				Description: "plugins initialized",
+			}
+			pie.Plugins = append(pie.Plugins, pluginData)
+			postInitPlugins(pie)
+
+		}, 3)
+
+		It("should correctly identify non-proper subsets with respect to maps", func() {
+
+			//test b proper subset of a
+			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+				[]common.Change{common.Change{Table: "b"}},
+			)).To(BeFalse())
+
+			//test a == b
+			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+				[]common.Change{common.Change{Table: "a"}, common.Change{Table: "b"}},
+			)).To(BeFalse())
+
+			//test b superset of a
+			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+				[]common.Change{common.Change{Table: "a"}, common.Change{Table: "b"}, common.Change{Table: "c"}},
+			)).To(BeTrue())
+
+			//test b not subset of a
+			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+				[]common.Change{common.Change{Table: "c"}},
+			)).To(BeTrue())
+
+			//test a empty
+			Expect(changesHaveNewTables(map[string]bool{},
+				[]common.Change{common.Change{Table: "a"}},
+			)).To(BeTrue())
+
+			//test b empty
+			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true},
+				[]common.Change{},
+			)).To(BeFalse())
+
+			//test b nil
+			Expect(changesHaveNewTables(map[string]bool{"a": true, "b": true}, nil)).To(BeTrue())
+
+			//test a nil
+			Expect(changesHaveNewTables(nil,
+				[]common.Change{common.Change{Table: "a"}},
+			)).To(BeTrue())
+		}, 3)
+
+		// todo: disabled for now -
+		// there is precondition I haven't been able to track down that breaks this test on occasion
+		XIt("should process a new snapshot when change server requires it", func(done Done) {
+			oldSnap := apidInfo.LastSnapshot
+			apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+				defer GinkgoRecover()
+
+				if s, ok := event.(*common.Snapshot); ok {
+					Expect(s.SnapshotInfo).NotTo(Equal(oldSnap))
+					close(done)
+				}
+			})
+			testMock.forceNewSnapshot()
 		})
 
-		bootstrap()
+
+		It("Verify the Sequence Number Logic works as expected", func() {
+			Expect(getChangeStatus("1.1.1", "1.1.2")).To(Equal(1))
+			Expect(getChangeStatus("1.1.1", "1.2.1")).To(Equal(1))
+			Expect(getChangeStatus("1.2.1", "1.2.1")).To(Equal(0))
+			Expect(getChangeStatus("1.2.1", "1.2.2")).To(Equal(1))
+			Expect(getChangeStatus("2.2.1", "1.2.2")).To(Equal(-1))
+			Expect(getChangeStatus("2.2.1", "2.2.0")).To(Equal(-1))
+		}, 3)
+
+		/*
+		 * XAPID-869, there should not be any panic if received duplicate snapshots during bootstrap
+		 */
+		It("Should be able to handle duplicate snapshot during bootstrap", func() {
+			initializeContext()
+
+			tokenManager = createTokenManager()
+			events.Listen(ApigeeSyncEventSelector, &handler{})
+
+			scopes := []string{apidInfo.ClusterID}
+			snapshot := &common.Snapshot{}
+			downloadSnapshot(scopes, snapshot, nil)
+			storeBootSnapshot(snapshot)
+			storeDataSnapshot(snapshot)
+			restoreContext()
+		}, 3)
 	})
-
-	It("should correctly identify non-proper subsets with respect to maps", func() {
-
-		//test b proper subset of a
-		Expect(mapIsSubset(map[string]bool{"a": true, "b": true}, map[string]bool{"b": true})).To(BeTrue())
-
-		//test a == b
-		Expect(mapIsSubset(map[string]bool{"a": true, "b": true}, map[string]bool{"a": true, "b": true})).To(BeTrue())
-
-		//test b superset of a
-		Expect(mapIsSubset(map[string]bool{"a": true, "b": true}, map[string]bool{"a": true, "b": true, "c": true})).To(BeFalse())
-
-		//test b not subset of a
-		Expect(mapIsSubset(map[string]bool{"a": true, "b": true}, map[string]bool{"c": true})).To(BeFalse())
-
-		//test b empty
-		Expect(mapIsSubset(map[string]bool{"a": true, "b": true}, map[string]bool{})).To(BeTrue())
-
-		//test a empty
-		Expect(mapIsSubset(map[string]bool{}, map[string]bool{"b": true})).To(BeFalse())
-	})
-
-	// todo: disabled for now -
-	// there is precondition I haven't been able to track down that breaks this test on occasion
-	XIt("should process a new snapshot when change server requires it", func(done Done) {
-		oldSnap := apidInfo.LastSnapshot
-		apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
-			defer GinkgoRecover()
-
-			if s, ok := event.(*common.Snapshot); ok {
-				Expect(s.SnapshotInfo).NotTo(Equal(oldSnap))
-				close(done)
-			}
-		})
-		testMock.forceNewSnapshot()
-	})
-
-	It("Verify the Sequence Number Logic works as expected", func() {
-		Expect(getChangeStatus("1.1.1", "1.1.2")).To(Equal(1))
-		Expect(getChangeStatus("1.1.1", "1.2.1")).To(Equal(1))
-		Expect(getChangeStatus("1.2.1", "1.2.1")).To(Equal(0))
-		Expect(getChangeStatus("1.2.1", "1.2.2")).To(Equal(1))
-		Expect(getChangeStatus("2.2.1", "1.2.2")).To(Equal(-1))
-		Expect(getChangeStatus("2.2.1", "2.2.0")).To(Equal(-1))
-	})
-
-	/*
-	 * XAPID-869, there should not be any panic if received duplicate snapshots during bootstrap
-	 */
-	It("Should be able to handle duplicate snapshot during bootstrap", func() {
-		scopes := []string{apidInfo.ClusterID}
-		snapshot := downloadSnapshot(scopes)
-		storeBootSnapshot(snapshot)
-		storeDataSnapshot(snapshot)
-	})
-
 })
diff --git a/backoff.go b/backoff.go
new file mode 100644
index 0000000..e3a7403
--- /dev/null
+++ b/backoff.go
@@ -0,0 +1,84 @@
+package apidApigeeSync
+
+import (
+	"math"
+	"time"
+	"math/rand"
+)
+
+const defaultInitial time.Duration = 200 * time.Millisecond
+const defaultMax time.Duration = 10 * time.Second
+const defaultFactor float64 = 2
+
+type Backoff struct {
+	attempt         int
+	initial, max    time.Duration
+	jitter bool
+	backoffStrategy func() time.Duration
+}
+
+type ExponentialBackoff struct {
+	Backoff
+	factor float64
+}
+
+func NewExponentialBackoff(initial, max time.Duration, factor float64, jitter bool) *ExponentialBackoff {
+	backoff := &ExponentialBackoff{}
+
+	if initial <= 0 {
+		initial = defaultInitial
+	}
+	if max <= 0 {
+		max = defaultMax
+	}
+
+	if factor <= 0 {
+		factor = defaultFactor
+	}
+
+	backoff.initial = initial
+	backoff.max = max
+	backoff.attempt = 0
+	backoff.factor = factor
+	backoff.jitter = jitter
+	backoff.backoffStrategy = backoff.exponentialBackoffStrategy
+
+	return backoff
+}
+
+func (b *Backoff) Duration() time.Duration {
+	d := b.backoffStrategy()
+	b.attempt++
+	return d
+}
+
+func (b *ExponentialBackoff) exponentialBackoffStrategy() time.Duration {
+
+	initial := float64(b.Backoff.initial)
+	attempt := float64(b.Backoff.attempt)
+	duration := initial * math.Pow(b.factor, attempt)
+
+	if duration > math.MaxInt64 {
+		return b.max
+	}
+	dur := time.Duration(duration)
+
+	if b.jitter {
+		duration = (rand.Float64()*(duration-initial) + initial)
+	}
+
+	if dur > b.max {
+		return b.max
+	}
+
+	log.Debugf("Backing off for %d ms", int64(dur/time.Millisecond))
+	return dur
+}
+
+func (b *Backoff) Reset() {
+	b.attempt = 0
+}
+
+func (b *Backoff) Attempt() int {
+	return b.attempt
+}
diff --git a/backoff_test.go b/backoff_test.go
new file mode 100644
index 0000000..ae85909
--- /dev/null
+++ b/backoff_test.go
@@ -0,0 +1,56 @@
+package apidApigeeSync
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"time"
+)
+
+var _ = Describe("backoff", func() {
+
+	Context("Backoff timeout calculations", func() {
+
+		It("Should properly apply defaults", func() {
+			log.Info("Starting backoff tests...")
+			b := NewExponentialBackoff(0, 0, 0, true)
+			Expect(defaultInitial).To(Equal(b.initial))
+			Expect(defaultMax).To(Equal(b.max))
+			Expect(defaultFactor).To(Equal(b.factor))
+
+			b = NewExponentialBackoff(-1, -1, -1, true)
+			Expect(defaultInitial).To(Equal(b.initial))
+			Expect(defaultMax).To(Equal(b.max))
+			Expect(defaultFactor).To(Equal(b.factor))
+		})
+
+		It("should properly apply exponential backoff strategy", func() {
+			b := NewExponentialBackoff(200 * time.Millisecond, 2 * time.Second, 2, false)
+			Expect(200 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(1).To(Equal(b.Attempt()))
+			Expect(400 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(2).To(Equal(b.Attempt()))
+			Expect(800 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(3).To(Equal(b.Attempt()))
+			Expect(1600 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(4).To(Equal(b.Attempt()))
+		})
+
+		It("should reset properly", func() {
+			b := NewExponentialBackoff(200 * time.Millisecond, 2 * time.Second, 2, false)
+			Expect(200 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(1).To(Equal(b.Attempt()))
+			Expect(400 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(2).To(Equal(b.Attempt()))
+			Expect(800 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(3).To(Equal(b.Attempt()))
+			b.Reset()
+			Expect(200 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(1).To(Equal(b.Attempt()))
+			Expect(400 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(2).To(Equal(b.Attempt()))
+			Expect(800 * time.Millisecond).To(Equal(b.Duration()))
+			Expect(3).To(Equal(b.Attempt()))
+		})
+	})
+
+})
diff --git a/changes.go b/changes.go
new file mode 100644
index 0000000..aa8d822
--- /dev/null
+++ b/changes.go
@@ -0,0 +1,319 @@
+package apidApigeeSync
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+
+	"github.com/apigee-labs/transicator/common"
+	"sync/atomic"
+)
+
+var lastSequence string
+var block string = "45"
+
+type pollChangeManager struct {
+	// 0 for not closed, 1 for closed
+	isClosed *int32
+	// 0 for pollChangeWithBackoff() not launched, 1 for launched
+	isLaunched *int32
+	quitChan   chan bool
+}
+
+func createChangeManager() *pollChangeManager {
+	isClosedInt := int32(0)
+	isLaunchedInt := int32(0)
+	return &pollChangeManager{
+		isClosed:   &isClosedInt,
+		quitChan:   make(chan bool),
+		isLaunched: &isLaunchedInt,
+	}
+}
+
+/*
+ * thread-safe close of pollChangeManager
+ * It marks status as closed immediately, quits backoff polling agent, and closes tokenManager
+ * use <- close() for blocking close
+ */
+func (c *pollChangeManager) close() <-chan bool {
+	finishChan := make(chan bool, 1)
+	//has been closed
+	if atomic.SwapInt32(c.isClosed, 1) == int32(1) {
+		log.Error("pollChangeManager: close() called on a closed pollChangeManager!")
+		go func() {
+			finishChan <- false
+			log.Debug("change manager closed")
+		}()
+		return finishChan
+	}
+	// not launched
+	if atomic.LoadInt32(c.isLaunched) == int32(0) {
+		log.Error("pollChangeManager: close() called when pollChangeWithBackoff unlaunched! close tokenManager!")
+		go func() {
+			tokenManager.close()
+			finishChan <- false
+			log.Debug("change manager closed")
+		}()
+		return finishChan
+	}
+	// launched
+	log.Debug("pollChangeManager: close pollChangeWithBackoff and token manager")
+	go func() {
+		c.quitChan <- true
+		tokenManager.close()
+		finishChan <- true
+		log.Debug("change manager closed")
+	}()
+	return finishChan
+}
+
+/*
+ * thread-safe pollChangeWithBackoff(), guaranteed: only one polling thread
+ */
+
+func (c *pollChangeManager) pollChangeWithBackoff() {
+	// closed
+	if atomic.LoadInt32(c.isClosed) == int32(1) {
+		log.Error("pollChangeManager: pollChangeWithBackoff() called after closed")
+		return
+	}
+	// has been launched before
+	if atomic.SwapInt32(c.isLaunched, 1) == int32(1) {
+		log.Error("pollChangeManager: pollChangeWithBackoff() has been launched before")
+		return
+	}
+
+	go pollWithBackoff(c.quitChan, c.pollChangeAgent, c.handleChangeServerError)
+	log.Debug("pollChangeManager: pollChangeWithBackoff() started pollWithBackoff")
+
+}
+
+/*
+ * Long polls the change agent with a 45 second block. Parses the response from
+ * change agent and raises an event. Called by pollWithBackoff().
+ */
+func (c *pollChangeManager) pollChangeAgent(dummyQuit chan bool) error {
+
+	changesUri, err := url.Parse(config.GetString(configChangeServerBaseURI))
+	if err != nil {
+		log.Errorf("bad url value for config %s: %s", changesUri, err)
+		return err
+	}
+	changesUri.Path = path.Join(changesUri.Path, "changes")
+
+	/*
+	 * Check to see if we have lastSequence already saved in the DB,
+	 * in which case, it has to be used to prevent re-reading same data
+	 */
+	lastSequence = getLastSequence()
+
+	for {
+		select {
+		case <-c.quitChan:
+			log.Info("pollChangeAgent; Recevied quit signal to stop polling change server, close token manager")
+			return quitSignalError{}
+		default:
+			err := c.getChanges(changesUri)
+			if err != nil {
+				if _, ok := err.(quitSignalError); ok {
+					log.Debug("pollChangeAgent: consuming the quit signal")
+					<-c.quitChan
+				}
+				return err
+			}
+		}
+	}
+}
+
+//TODO refactor this method more, split it up
+/* Make a single request to the changeserver to get a changelist */
+func (c *pollChangeManager) getChanges(changesUri *url.URL) error {
+	// if closed
+	if atomic.LoadInt32(c.isClosed) == int32(1) {
+		return quitSignalError{}
+	}
+	log.Debug("polling...")
+
+	/* Find the scopes associated with the config id */
+	scopes := findScopesForId(apidInfo.ClusterID)
+	v := url.Values{}
+
+	/* Sequence added to the query if available */
+	if lastSequence != "" {
+		v.Add("since", lastSequence)
+	}
+	v.Add("block", block)
+
+	/*
+	 * Include all the scopes associated with the config Id
+	 * The Config Id is included as well, as it acts as the
+	 * Bootstrap scope
+	 */
+	for _, scope := range scopes {
+		v.Add("scope", scope)
+	}
+	v.Add("scope", apidInfo.ClusterID)
+	v.Add("snapshot", apidInfo.LastSnapshot)
+	changesUri.RawQuery = v.Encode()
+	uri := changesUri.String()
+	log.Debugf("Fetching changes: %s", uri)
+
+	/* If error, break the loop, and retry after interval */
+	client := &http.Client{Timeout: httpTimeout} // must be greater than block value
+	req, err := http.NewRequest("GET", uri, nil)
+	addHeaders(req)
+
+	r, err := client.Do(req)
+	if err != nil {
+		log.Errorf("change agent comm error: %s", err)
+		// if closed
+		if atomic.LoadInt32(c.isClosed) == int32(1) {
+			return quitSignalError{}
+		}
+		return err
+	}
+	defer r.Body.Close()
+
+	// has been closed
+	if atomic.LoadInt32(c.isClosed) == int32(1) {
+		log.Debugf("getChanges: changeManager has been closed")
+		return quitSignalError{}
+	}
+
+	if r.StatusCode != http.StatusOK {
+		log.Errorf("Get changes request failed with status code: %d", r.StatusCode)
+		switch r.StatusCode {
+		case http.StatusUnauthorized:
+			tokenManager.invalidateToken()
+			return nil
+
+		case http.StatusNotModified:
+			return nil
+
+		case http.StatusBadRequest:
+			var apiErr changeServerError
+			var b []byte
+			b, err = ioutil.ReadAll(r.Body)
+			if err != nil {
+				log.Errorf("Unable to read response body: %v", err)
+				return err
+			}
+			err = json.Unmarshal(b, &apiErr)
+			if err != nil {
+				log.Errorf("JSON Response Data not parsable: %s", string(b))
+				return err
+			}
+			if apiErr.Code == "SNAPSHOT_TOO_OLD" {
+				log.Debug("Received SNAPSHOT_TOO_OLD message from change server.")
+				err = apiErr
+			}
+			return nil
+		}
+		return nil
+	}
+
+	var resp common.ChangeList
+	err = json.NewDecoder(r.Body).Decode(&resp)
+	if err != nil {
+		log.Errorf("JSON Response Data not parsable: %v", err)
+		return err
+	}
+
+	/*
+	 * If the lastSequence is already newer or the same than what we got via
+	 * resp.LastSequence, Ignore it.
+	 */
+	if lastSequence != "" &&
+		getChangeStatus(lastSequence, resp.LastSequence) != 1 {
+		return changeServerError{
+			Code: "Ignore change, already have newer changes",
+		}
+	}
+
+	if changesRequireDDLSync(resp) {
+		return changeServerError{
+			Code: "DDL changes detected; must get new snapshot",
+		}
+	}
+
+	/* If valid data present, Emit to plugins */
+	if len(resp.Changes) > 0 {
+		select {
+		case <-time.After(httpTimeout):
+			log.Panic("Timeout. Plugins failed to respond to changes.")
+		case <-events.Emit(ApigeeSyncEventSelector, &resp):
+		}
+	} else {
+		log.Debugf("No Changes detected for Scopes: %s", scopes)
+	}
+
+	updateSequence(resp.LastSequence)
+
+	return nil
+}
+
+func changesRequireDDLSync(changes common.ChangeList) bool {
+	return changesHaveNewTables(knownTables, changes.Changes)
+}
+
+func (c *pollChangeManager) handleChangeServerError(err error) {
+	// has been closed
+	if atomic.LoadInt32(c.isClosed) == int32(1) {
+		log.Debugf("handleChangeServerError: changeManager has been closed")
+		return
+	}
+	if _, ok := err.(changeServerError); ok {
+		log.Info("Detected DDL changes, going to fetch a new snapshot to sync...")
+		downloadDataSnapshot(c.quitChan)
+	} else {
+		log.Debugf("Error connecting to changeserver: %v", err)
+	}
+}
+
+/*
+ * Determine if any tables in changes are not present in known tables
+ */
+func changesHaveNewTables(a map[string]bool, changes []common.Change) bool {
+
+	//nil maps should not be passed in.  Making the distinction between nil map and empty map
+	if a == nil || changes == nil {
+		return true
+	}
+
+	for _, change := range changes {
+		if !a[change.Table] {
+			log.Infof("Unable to find %s table in current known tables", change.Table)
+			return true
+		}
+	}
+
+	return false
+}
+
+/*
+ * seqCurr.Compare() will return 1, if its newer than seqPrev,
+ * else will return 0, if same, or -1 if older.
+ */
+func getChangeStatus(lastSeq string, currSeq string) int {
+	seqPrev, err := common.ParseSequence(lastSeq)
+	if err != nil {
+		log.Panic("Unable to parse previous sequence string")
+	}
+	seqCurr, err := common.ParseSequence(currSeq)
+	if err != nil {
+		log.Panic("Unable to parse current sequence string")
+	}
+	return seqCurr.Compare(seqPrev)
+}
+
+func updateSequence(seq string) {
+	lastSequence = seq
+	err := updateLastSequence(seq)
+	if err != nil {
+		log.Panic("Unable to update Sequence in DB")
+	}
+
+}
diff --git a/data.go b/data.go
index 08ff3ba..48f1ff7 100644
--- a/data.go
+++ b/data.go
@@ -88,8 +88,9 @@
 
 	log.Debugf("inserting into APID_CLUSTER: %v", dac)
 
+	//replace to accomodate same snapshot txid
 	stmt, err := txn.Prepare(`
-	INSERT INTO APID_CLUSTER
+	REPLACE INTO APID_CLUSTER
 		(id, description, name, umbrella_org_app_name,
 		created, created_by, updated, updated_by,
 		last_sequence)
@@ -117,8 +118,9 @@
 
 	log.Debugf("insert DATA_SCOPE: %v", ds)
 
+	//replace to accomodate same snapshot txid
 	stmt, err := txn.Prepare(`
-	INSERT INTO DATA_SCOPE
+	REPLACE INTO DATA_SCOPE
 		(id, apid_cluster_id, scope, org,
 		env, created, created_by, updated,
 		updated_by)
@@ -237,7 +239,7 @@
 
 	// always use default database for this
 	var db apid.DB
-	db, err = data.DB()
+	db, err = dataService.DB()
 	if err != nil {
 		return
 	}
@@ -254,6 +256,7 @@
 			newInstanceID = true
 			info.InstanceID = generateUUID()
 
+			log.Debugf("Inserting new apid instance id %s", info.InstanceID)
 			db.Exec("INSERT INTO APID (instance_id, last_snapshot_info) VALUES (?,?)",
 				info.InstanceID, "")
 		}
@@ -264,7 +267,7 @@
 func updateApidInstanceInfo() error {
 
 	// always use default database for this
-	db, err := data.DB()
+	db, err := dataService.DB()
 	if err != nil {
 		return err
 	}
diff --git a/datascope_cache.go b/datascope_cache.go
deleted file mode 100644
index 3d19710..0000000
--- a/datascope_cache.go
+++ /dev/null
@@ -1,93 +0,0 @@
-package apidApigeeSync
-
-const (
-	readCache int = iota
-	updateCache
-	removeCache
-	clearAndInit
-)
-
-/*
- * structs for DatascopeCache
- */
-
-type cacheOperationRequest struct {
-	Operation int
-	Scope     *dataDataScope
-	version   string
-}
-
-// maintain an in-mem cache of datascope
-type DatascopeCache struct {
-	requestChan  chan *cacheOperationRequest
-	readDoneChan chan []string
-	scopeMap     map[string]*dataDataScope
-	version      string
-}
-
-var scopeCache *DatascopeCache
-
-func (cache *DatascopeCache) datascopeCacheManager() {
-	for request := range cache.requestChan {
-		switch request.Operation {
-		case readCache:
-			log.Debug("datascopeCacheManager: readCache")
-			scopes := make([]string, 0, len(cache.scopeMap))
-			for _, ds := range cache.scopeMap {
-				scopes = append(scopes, ds.Scope)
-			}
-			cache.readDoneChan <- scopes
-		case updateCache:
-			log.Debug("datascopeCacheManager: updateCache")
-			cache.scopeMap[request.Scope.ID] = request.Scope
-		case removeCache:
-			log.Debug("datascopeCacheManager: removeCache")
-			delete(cache.scopeMap, request.Scope.ID)
-		case clearAndInit:
-			log.Debug("datascopeCacheManager: clearAndInit")
-			if cache.version != request.version {
-				cache.scopeMap = make(map[string]*dataDataScope)
-				cache.version = request.version
-			}
-		}
-	}
-
-	//chan closed
-	cache.scopeMap = nil
-	close(cache.readDoneChan)
-}
-
-/*
- * The output of readAllScope() should be identical to findScopesForId(apidInfo.ClusterID)
- */
-
-func (cache *DatascopeCache) readAllScope() []string {
-	cache.requestChan <- &cacheOperationRequest{readCache, nil, ""}
-	scopes := <-cache.readDoneChan
-	// eliminate duplicates
-	tmpMap := make(map[string]bool)
-	for _, scope := range scopes {
-		tmpMap[scope] = true
-	}
-	scopes = make([]string, 0)
-	for scope := range tmpMap {
-		scopes = append(scopes, scope)
-	}
-	return scopes
-}
-
-func (cache *DatascopeCache) removeCache(scope *dataDataScope) {
-	cache.requestChan <- &cacheOperationRequest{removeCache, scope, ""}
-}
-
-func (cache *DatascopeCache) updateCache(scope *dataDataScope) {
-	cache.requestChan <- &cacheOperationRequest{updateCache, scope, ""}
-}
-
-func (cache *DatascopeCache) clearAndInitCache(version string) {
-	cache.requestChan <- &cacheOperationRequest{clearAndInit, nil, version}
-}
-
-func (cache *DatascopeCache) closeCache() {
-	close(cache.requestChan)
-}
diff --git a/datascope_cache_test.go b/datascope_cache_test.go
deleted file mode 100644
index bc67f53..0000000
--- a/datascope_cache_test.go
+++ /dev/null
@@ -1,97 +0,0 @@
-package apidApigeeSync
-
-import (
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-	"math/rand"
-	"strconv"
-	"time"
-)
-
-var _ = Describe("datascope cache", func() {
-	/*
-	 * in-mem cache test
-	 */
-	It("Test In-mem cache", func() {
-		testCache := &DatascopeCache{requestChan: make(chan *cacheOperationRequest), readDoneChan: make(chan []string)}
-		go testCache.datascopeCacheManager()
-		testCache.clearAndInitCache("test-version")
-		countChan := make(chan int)
-		base := 10
-		rand.Seed(time.Now().Unix())
-		num := base + rand.Intn(base)
-		scopeMap := make(map[string]bool)
-		// async update
-		for i := 0; i < num; i++ {
-			id := strconv.Itoa(i)
-			scopeStr := strconv.Itoa(i % base)
-			scope := &dataDataScope{ID: id, Scope: scopeStr}
-			scopeMap[scope.Scope] = true
-			go func(scope *dataDataScope) {
-				testCache.updateCache(scope)
-				countChan <- 1
-			}(scope)
-		}
-
-		// wait until update done
-		for i := 0; i < num; i++ {
-			<-countChan
-		}
-
-		// verify update
-		retrievedScopes := testCache.readAllScope()
-		Expect(len(scopeMap)).To(Equal(len(retrievedScopes)))
-		for _, s := range retrievedScopes {
-			// verify each retrieved scope is valid
-			Expect(scopeMap[s]).To(BeTrue())
-			// no duplicate scopes
-			scopeMap[s] = true
-		}
-
-		// remove all the datascopes with odd scope
-		count := 0
-		for i := 0; i < num; i++ {
-			if (i%base)%2 == 1 {
-				count += 1
-				id := strconv.Itoa(i)
-				scopeStr := strconv.Itoa(i % base)
-				scope := &dataDataScope{ID: id, Scope: scopeStr}
-				go func(scope *dataDataScope) {
-					testCache.removeCache(scope)
-					countChan <- 1
-				}(scope)
-			}
-		}
-
-		for i := 0; i < count; i++ {
-			<-countChan
-		}
-
-		// all retrieved scopes should be even
-		retrievedScopes = testCache.readAllScope()
-		for _, s := range retrievedScopes {
-			scopeNum, _ := strconv.Atoi(s)
-			Expect(scopeNum % 2).To(BeZero())
-		}
-
-		// async remove all datascopes
-		for i := 0; i < num; i++ {
-			id := strconv.Itoa(i)
-			scopeStr := strconv.Itoa(i % base)
-			scope := &dataDataScope{ID: id, Scope: scopeStr}
-			go func(scope *dataDataScope) {
-				testCache.removeCache(scope)
-				countChan <- 1
-			}(scope)
-		}
-
-		for i := 0; i < num; i++ {
-			<-countChan
-		}
-		retrievedScopes = testCache.readAllScope()
-		Expect(len(retrievedScopes)).To(Equal(0))
-
-		testCache.closeCache()
-	})
-
-})
diff --git a/init.go b/init.go
index d40a9e7..e1b31c5 100644
--- a/init.go
+++ b/init.go
@@ -29,14 +29,21 @@
 )
 
 var (
-	log               apid.LogService
-	config            apid.ConfigService
-	data              apid.DataService
-	events            apid.EventsService
-	apidInfo          apidInstanceInfo
-	apidPluginDetails string
-	newInstanceID     bool
-	tokenManager      *tokenMan
+	/* All set during plugin initialization */
+	log                       apid.LogService
+	config                    apid.ConfigService
+	dataService               apid.DataService
+	events                    apid.EventsService
+	apidInfo                  apidInstanceInfo
+	newInstanceID             bool
+	tokenManager              *tokenMan
+	changeManager             *pollChangeManager
+	quitPollingSnapshotServer chan bool
+
+	/* Set during post plugin initialization
+	 * set this as a default, so that it's guaranteed to be valid even if postInitPlugins isn't called
+	 */
+	apidPluginDetails string = `[{"name":"apidApigeeSync","schemaVer":"1.0"}]`
 )
 
 type apidInstanceInfo struct {
@@ -52,7 +59,7 @@
 	apid.RegisterPlugin(initPlugin)
 }
 
-func initDefaults() {
+func initConfigDefaults() {
 	config.SetDefault(configPollInterval, 120*time.Second)
 	config.SetDefault(configSnapshotProtocol, "json")
 	name, errh := os.Hostname()
@@ -64,57 +71,28 @@
 	log.Debugf("Using %s as display name", config.GetString(configName))
 }
 
-func SetLogger(logger apid.LogService) {
-	log = logger
-}
-
-func initPlugin(services apid.Services) (apid.PluginData, error) {
-	SetLogger(services.Log().ForModule("apigeeSync"))
-	log.Debug("start init")
-
-	config = services.Config()
-	initDefaults()
-
-	data = services.Data()
+func initVariables(services apid.Services) error {
+	dataService = services.Data()
 	events = services.Events()
-
-	scopeCache = &DatascopeCache{requestChan: make(chan *cacheOperationRequest), readDoneChan: make(chan []string)}
-
-	go scopeCache.datascopeCacheManager()
-
-	/* This callback function will get called, once all the plugins are
-	 * initialized (not just this plugin). This is needed because,
-	 * downloadSnapshots/changes etc have to begin to be processed only
-	 * after all the plugins are initialized
-	 */
-	events.ListenOnceFunc(apid.SystemEventsSelector, postInitPlugins)
-
-	// check for required values
-	for _, key := range []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret,
-		configSnapServerBaseURI, configChangeServerBaseURI} {
-		if !config.IsSet(key) {
-			return pluginData, fmt.Errorf("Missing required config value: %s", key)
-		}
-	}
-	proto := config.GetString(configSnapshotProtocol)
-	if proto != "json" && proto != "proto" {
-		return pluginData, fmt.Errorf("Illegal value for %s. Must be: 'json' or 'proto'", configSnapshotProtocol)
-	}
+	//TODO listen for arbitrary commands, these channels can be used to kill polling goroutines
+	//also useful for testing
+	quitPollingSnapshotServer = make(chan bool)
+	changeManager = createChangeManager()
 
 	// set up default database
-	db, err := data.DB()
+	db, err := dataService.DB()
 	if err != nil {
-		return pluginData, fmt.Errorf("Unable to access DB: %v", err)
+		return fmt.Errorf("Unable to access DB: %v", err)
 	}
 	err = initDB(db)
 	if err != nil {
-		return pluginData, fmt.Errorf("Unable to access DB: %v", err)
+		return fmt.Errorf("Unable to access DB: %v", err)
 	}
 	setDB(db)
 
 	apidInfo, err = getApidInstanceInfo()
 	if err != nil {
-		return pluginData, fmt.Errorf("Unable to get apid instance info: %v", err)
+		return fmt.Errorf("Unable to get apid instance info: %v", err)
 	}
 
 	if config.IsSet(configApidInstanceID) {
@@ -122,6 +100,64 @@
 	}
 	config.Set(configApidInstanceID, apidInfo.InstanceID)
 
+	return nil
+}
+
+func checkForRequiredValues() error {
+	// check for required values
+	for _, key := range []string{configProxyServerBaseURI, configConsumerKey, configConsumerSecret,
+		configSnapServerBaseURI, configChangeServerBaseURI} {
+		if !config.IsSet(key) {
+			return fmt.Errorf("Missing required config value: %s", key)
+		}
+	}
+	proto := config.GetString(configSnapshotProtocol)
+	if proto != "json" && proto != "proto" {
+		return fmt.Errorf("Illegal value for %s. Must be: 'json' or 'proto'", configSnapshotProtocol)
+	}
+
+	return nil
+}
+
+func SetLogger(logger apid.LogService) {
+	log = logger
+}
+
+/* Idempotent state initialization */
+func _initPlugin(services apid.Services) error {
+	SetLogger(services.Log().ForModule("apigeeSync"))
+	log.Debug("start init")
+
+	config = services.Config()
+	err := checkForRequiredValues()
+	if err != nil {
+		return err
+	}
+
+	initConfigDefaults()
+
+	err = initVariables(services)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func initPlugin(services apid.Services) (apid.PluginData, error) {
+
+	err := _initPlugin(services)
+	if err != nil {
+		return pluginData, err
+	}
+
+	/* This callback function will get called once all the plugins are
+	 * initialized (not just this plugin). This is needed because,
+	 * downloadSnapshots/changes etc have to begin to be processed only
+	 * after all the plugins are initialized
+	 */
+	events.ListenOnceFunc(apid.SystemEventsSelector, postInitPlugins)
+
 	log.Debug("end init")
 
 	return pluginData, nil
diff --git a/init_test.go b/init_test.go
index 7bcee56..479b4b6 100644
--- a/init_test.go
+++ b/init_test.go
@@ -10,18 +10,20 @@
 	Context("Apid Instance display name", func() {
 
 		It("should be hostname by default", func() {
-			initDefaults()
+			log.Info("Starting init tests...")
+
+			initConfigDefaults()
 			Expect(apidInfo.InstanceName).To(Equal("testhost"))
-		})
+		}, 3)
 
 		It("accept display name from config", func() {
 			config.Set(configName, "aa01")
-			initDefaults()
+			initConfigDefaults()
 			var apidInfoLatest apidInstanceInfo
 			apidInfoLatest, _ = getApidInstanceInfo()
 			Expect(apidInfoLatest.InstanceName).To(Equal("aa01"))
 			Expect(apidInfoLatest.LastSnapshot).To(Equal(""))
-		})
+		}, 3)
 
 	})
 
diff --git a/listener.go b/listener.go
index 1fbd82e..6c4b1ef 100644
--- a/listener.go
+++ b/listener.go
@@ -29,22 +29,42 @@
 }
 
 func processSnapshot(snapshot *common.Snapshot) {
-
 	log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo)
 
-	db, err := data.DBVersion(snapshot.SnapshotInfo)
+	db, err := dataService.DBVersion(snapshot.SnapshotInfo)
 	if err != nil {
 		log.Panicf("Unable to access database: %v", err)
 	}
 
-	err = initDB(db)
+	if config.GetString(configSnapshotProtocol) == "json" {
+		processJsonSnapshot(snapshot, db)
+	} else if config.GetString(configSnapshotProtocol) == "sqlite" {
+		processSqliteSnapshot(snapshot, db)
+	}
+
+	//update apid instance info
+	apidInfo.LastSnapshot = snapshot.SnapshotInfo
+	err = updateApidInstanceInfo()
+	if err != nil {
+		log.Panicf("Unable to update instance info: %v", err)
+	}
+
+	setDB(db)
+	log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo)
+
+}
+
+func processSqliteSnapshot(snapshot *common.Snapshot, db apid.DB) {
+	//nothing to do as of now, here as a placeholder
+}
+
+func processJsonSnapshot(snapshot *common.Snapshot, db apid.DB) {
+
+	err := initDB(db)
 	if err != nil {
 		log.Panicf("Unable to initialize database: %v", err)
 	}
 
-	// clear cache
-	scopeCache.clearAndInitCache(snapshot.SnapshotInfo)
-
 	tx, err := db.Begin()
 	if err != nil {
 		log.Panicf("Error starting transaction: %v", err)
@@ -73,10 +93,6 @@
 				if err != nil {
 					log.Panicf("Snapshot update failed: %v", err)
 				}
-				// cache scopes for this cluster
-				if ds.ClusterID == apidInfo.ClusterID {
-					scopeCache.updateCache(&ds)
-				}
 			}
 		}
 	}
@@ -85,15 +101,6 @@
 	if err != nil {
 		log.Panicf("Error committing Snapshot change: %v", err)
 	}
-
-	apidInfo.LastSnapshot = snapshot.SnapshotInfo
-	err = updateApidInstanceInfo()
-	if err != nil {
-		log.Panicf("Unable to update instance info: %v", err)
-	}
-
-	setDB(db)
-	log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo)
 }
 
 func processChangeList(changes *common.ChangeList) {
@@ -121,19 +128,9 @@
 			case common.Insert:
 				ds := makeDataScopeFromRow(change.NewRow)
 				err = insertDataScope(ds, tx)
-
-				// cache scopes for this cluster
-				if (ds.ClusterID == apidInfo.ClusterID) && (err == nil) {
-					scopeCache.updateCache(&ds)
-				}
 			case common.Delete:
 				ds := makeDataScopeFromRow(change.OldRow)
 				err = deleteDataScope(ds, tx)
-
-				// cache scopes for this cluster
-				if (ds.ClusterID == apidInfo.ClusterID) && (err == nil) {
-					scopeCache.removeCache(&ds)
-				}
 			default:
 				// common.Update is not allowed
 				log.Panicf("illegal operation: %s for %s", change.Operation, change.Table)
diff --git a/listener_test.go b/listener_test.go
index 2b060de..ee0b4cc 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -15,6 +15,7 @@
 	Context("ApigeeSync snapshot event", func() {
 
 		It("should set DB to appropriate version", func() {
+			log.Info("Starting listener tests...")
 
 			//save the last snapshot, so we can restore it at the end of this context
 			saveLastSnapshot = apidInfo.LastSnapshot
@@ -28,7 +29,7 @@
 
 			Expect(apidInfo.LastSnapshot).To(Equal(event.SnapshotInfo))
 
-			expectedDB, err := data.DBVersion(event.SnapshotInfo)
+			expectedDB, err := dataService.DBVersion(event.SnapshotInfo)
 			Expect(err).NotTo(HaveOccurred())
 
 			Expect(getDB() == expectedDB).Should(BeTrue())
@@ -47,7 +48,7 @@
 			}
 
 			Expect(func() { handler.Handle(&event) }).To(Panic())
-		})
+		}, 3)
 
 		It("should process a valid Snapshot", func() {
 
@@ -203,7 +204,7 @@
 
 			//restore the last snapshot
 			apidInfo.LastSnapshot = saveLastSnapshot
-		})
+		}, 3)
 	})
 
 	Context("ApigeeSync change event", func() {
@@ -225,7 +226,7 @@
 				}
 
 				Expect(func() { handler.Handle(&event) }).To(Panic())
-			})
+			}, 3)
 
 			It("update event should panic", func() {
 
@@ -242,7 +243,7 @@
 				Expect(func() { handler.Handle(&event) }).To(Panic())
 				//restore the last snapshot
 				apidInfo.LastSnapshot = saveLastSnapshot
-			})
+			}, 3)
 
 			PIt("delete event should kill all the things!")
 		})
@@ -328,7 +329,7 @@
 				Expect(len(scopes)).To(Equal(2))
 				Expect(scopes[0]).To(Equal("s1"))
 				Expect(scopes[1]).To(Equal("s2"))
-			})
+			}, 3)
 
 			It("delete event should delete", func() {
 				insert := common.ChangeList{
@@ -372,7 +373,7 @@
 				Expect(err).NotTo(HaveOccurred())
 
 				Expect(nRows).To(Equal(0))
-			})
+			}, 3)
 
 			It("update event should panic", func() {
 
@@ -389,7 +390,7 @@
 				Expect(func() { handler.Handle(&event) }).To(Panic())
 				//restore the last snapshot
 				apidInfo.LastSnapshot = saveLastSnapshot
-			})
+			}, 3)
 
 		})
 
diff --git a/snapshot.go b/snapshot.go
new file mode 100644
index 0000000..ae667bf
--- /dev/null
+++ b/snapshot.go
@@ -0,0 +1,257 @@
+package apidApigeeSync
+
+import (
+	"encoding/json"
+	"github.com/30x/apid-core"
+	"github.com/30x/apid-core/data"
+	"github.com/apigee-labs/transicator/common"
+	"net/http"
+	"os"
+
+	"io"
+	"io/ioutil"
+	"net/url"
+	"path"
+	"time"
+)
+
+// retrieve boot information: apid_config and apid_config_scope
+func downloadBootSnapshot(quitPolling chan bool) {
+	log.Debug("download Snapshot for boot data")
+
+	scopes := []string{apidInfo.ClusterID}
+	snapshot := &common.Snapshot{}
+	downloadSnapshot(scopes, snapshot, quitPolling)
+	storeBootSnapshot(snapshot)
+}
+
+func storeBootSnapshot(snapshot *common.Snapshot) {
+	processSnapshot(snapshot)
+}
+
+// use the scope IDs from the boot snapshot to get all the data associated with the scopes
+func downloadDataSnapshot(quitPolling chan bool) {
+	log.Debug("download Snapshot for data scopes")
+
+	scopes := findScopesForId(apidInfo.ClusterID)
+	scopes = append(scopes, apidInfo.ClusterID)
+	snapshot := &common.Snapshot{}
+	downloadSnapshot(scopes, snapshot, quitPolling)
+	storeDataSnapshot(snapshot)
+}
+
+func storeDataSnapshot(snapshot *common.Snapshot) {
+	knownTables = extractTablesFromSnapshot(snapshot)
+
+	db, err := dataService.DBVersion(snapshot.SnapshotInfo)
+	if err != nil {
+		log.Panicf("Database inaccessible: %v", err)
+	}
+	persistKnownTablesToDB(knownTables, db)
+
+	log.Info("Emitting Snapshot to plugins")
+
+	select {
+	case <-time.After(pluginTimeout):
+		log.Panic("Timeout. Plugins failed to respond to snapshot.")
+	case <-events.Emit(ApigeeSyncEventSelector, snapshot):
+	}
+
+}
+
+func extractTablesFromSnapshot(snapshot *common.Snapshot) (tables map[string]bool) {
+
+	tables = make(map[string]bool)
+
+	log.Debug("Extracting table names from snapshot")
+	if snapshot.Tables == nil {
+		//if this panic ever fires, it's a bug
+		log.Panicf("Attempt to extract known tables from snapshot without tables failed")
+	}
+
+	for _, table := range snapshot.Tables {
+		tables[table.Name] = true
+	}
+
+	return tables
+}
+
+func extractTablesFromDB(db apid.DB) (tables map[string]bool) {
+
+	tables = make(map[string]bool)
+
+	log.Debug("Extracting table names from existing DB")
+	rows, err := db.Query("SELECT name FROM _known_tables;")
+	defer rows.Close()
+
+	if err != nil {
+		log.Panicf("Error reading current set of tables: %v", err)
+	}
+
+	for rows.Next() {
+		var table string
+		if err := rows.Scan(&table); err != nil {
+			log.Panicf("Error reading current set of tables: %v", err)
+		}
+		log.Debugf("Table %s found in existing db", table)
+
+		tables[table] = true
+	}
+	return tables
+}
+
+// Skip Downloading snapshot if there is already a snapshot available from previous run
+func startOnLocalSnapshot(snapshot string) *common.Snapshot {
+	log.Infof("Starting on local snapshot: %s", snapshot)
+
+	// ensure DB version will be accessible on behalf of dependant plugins
+	db, err := dataService.DBVersion(snapshot)
+	if err != nil {
+		log.Panicf("Database inaccessible: %v", err)
+	}
+
+	knownTables = extractTablesFromDB(db)
+
+	// allow plugins (including this one) to start immediately on existing database
+	// Note: this MUST have no tables as that is used as an indicator
+	return &common.Snapshot{
+		SnapshotInfo: snapshot,
+	}
+}
+
+// will keep retrying with backoff until success
+func downloadSnapshot(scopes []string, snapshot *common.Snapshot, quitPolling chan bool) {
+
+	log.Debug("downloadSnapshot")
+
+	snapshotUri, err := url.Parse(config.GetString(configSnapServerBaseURI))
+	if err != nil {
+		log.Panicf("bad url value for config %s: %s", snapshotUri, err)
+	}
+
+	snapshotUri.Path = path.Join(snapshotUri.Path, "snapshots")
+
+	v := url.Values{}
+	for _, scope := range scopes {
+		v.Add("scope", scope)
+	}
+	snapshotUri.RawQuery = v.Encode()
+	uri := snapshotUri.String()
+	log.Infof("Snapshot Download: %s", uri)
+
+	client := &http.Client{
+		CheckRedirect: Redirect,
+		Timeout:       httpTimeout,
+	}
+
+	//pollWithBackoff only accepts function that accept a single quit channel
+	//to accomadate functions which need more parameters, wrap them in closures
+	attemptDownload := getAttemptDownloadClosure(client, snapshot, uri)
+
+	pollWithBackoff(quitPolling, attemptDownload, handleSnapshotServerError)
+}
+
+func getAttemptDownloadClosure(client *http.Client, snapshot *common.Snapshot, uri string) func(chan bool) error {
+	return func(_ chan bool) error {
+		req, err := http.NewRequest("GET", uri, nil)
+		if err != nil {
+			// should never happen, but if it does, it's unrecoverable anyway
+			log.Panicf("Snapshotserver comm error: %v", err)
+		}
+		addHeaders(req)
+
+		var processSnapshotResponse func(*http.Response, *common.Snapshot) error
+
+		// Set the transport protocol type based on conf file input
+		if config.GetString(configSnapshotProtocol) == "json" {
+			req.Header.Set("Accept", "application/json")
+			processSnapshotResponse = processSnapshotServerJsonResponse
+		} else if config.GetString(configSnapshotProtocol) == "sqlite" {
+			req.Header.Set("Accept", "application/transicator+sqlite")
+			processSnapshotResponse = processSnapshotServerFileResponse
+		}
+
+		// Issue the request to the snapshot server
+		r, err := client.Do(req)
+		if err != nil {
+			log.Errorf("Snapshotserver comm error: %v", err)
+			return err
+		}
+
+		defer r.Body.Close()
+
+		if r.StatusCode != 200 {
+			body, _ := ioutil.ReadAll(r.Body)
+			log.Errorf("Snapshot server conn failed with resp code %d, body: %s", r.StatusCode, string(body))
+			return expected200Error{}
+		}
+
+		// Decode the Snapshot server response
+		err = processSnapshotResponse(r, snapshot)
+		if err != nil {
+			log.Errorf("Response Data not parsable: %v", err)
+			return err
+		}
+
+		return nil
+	}
+}
+
+func persistKnownTablesToDB(tables map[string]bool, db apid.DB) {
+	log.Debugf("Inserting table names found in snapshot into db")
+
+	tx, err := db.Begin()
+	if err != nil {
+		log.Panicf("Error starting transaction: %v", err)
+	}
+	defer tx.Rollback()
+
+	_, err = tx.Exec("CREATE TABLE _known_tables (name text, PRIMARY KEY(name));")
+	if err != nil {
+		log.Panicf("Could not create _known_tables table: %s", err)
+	}
+
+	for name := range tables {
+		log.Debugf("Inserting %s into _known_tables", name)
+		_, err := tx.Exec("INSERT INTO _known_tables VALUES(?);", name)
+		if err != nil {
+			log.Panicf("Error encountered inserting into known tables ", err)
+		}
+
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		log.Panicf("Error committing transaction: %v", err)
+
+	}
+}
+
+func processSnapshotServerJsonResponse(r *http.Response, snapshot *common.Snapshot) error {
+	return json.NewDecoder(r.Body).Decode(snapshot)
+}
+
+func processSnapshotServerFileResponse(r *http.Response, snapshot *common.Snapshot) error {
+	dbId := r.Header.Get("Transicator-Snapshot-TXID")
+	out, err := os.Create(data.DBPath(dbId))
+	if err != nil {
+		return err
+	}
+	defer out.Close()
+
+	//stream respose to DB
+	_, err = io.Copy(out, r.Body)
+
+	if err != nil {
+		return err
+	}
+
+	snapshot.SnapshotInfo = dbId
+	//TODO get timestamp from transicator.  Not currently in response
+
+	return nil
+}
+
+func handleSnapshotServerError(err error) {
+	log.Debugf("Error connecting to snapshot server: %v", err)
+}
diff --git a/token.go b/token.go
index 60097c1..a6c118e 100644
--- a/token.go
+++ b/token.go
@@ -7,13 +7,12 @@
 	"net/http"
 	"net/url"
 	"path"
-	"sync"
+	"sync/atomic"
 	"time"
 )
 
 var (
 	refreshFloatTime = time.Minute
-	getTokenLock     sync.Mutex
 )
 
 /*
@@ -26,15 +25,34 @@
 */
 
 func createTokenManager() *tokenMan {
-	t := &tokenMan{}
-	t.doRefresh = make(chan bool, 1)
-	t.maintainToken()
+	isClosedInt := int32(0)
+
+	t := &tokenMan{
+		quitPollingForToken: make(chan bool, 1),
+		closed:              make(chan bool),
+		getTokenChan:        make(chan bool),
+		invalidateTokenChan: make(chan bool),
+		returnTokenChan:     make(chan *oauthToken),
+		invalidateDone:      make(chan bool),
+		isClosed:            &isClosedInt,
+	}
+
+	t.retrieveNewToken()
+	t.refreshTimer = time.After(t.token.refreshIn())
+	go t.maintainToken()
 	return t
 }
 
 type tokenMan struct {
-	token     *oauthToken
-	doRefresh chan bool
+	token               *oauthToken
+	isClosed            *int32
+	quitPollingForToken chan bool
+	closed              chan bool
+	getTokenChan        chan bool
+	invalidateTokenChan chan bool
+	refreshTimer        <-chan time.Time
+	returnTokenChan     chan *oauthToken
+	invalidateDone      chan bool
 }
 
 func (t *tokenMan) getBearerToken() string {
@@ -42,50 +60,63 @@
 }
 
 func (t *tokenMan) maintainToken() {
-	go func() {
-		for {
-			if t.token.needsRefresh() {
-				getTokenLock.Lock()
-				t.retrieveNewToken()
-				getTokenLock.Unlock()
-			}
-			select {
-			case _, ok := <-t.doRefresh:
-				if !ok {
-					log.Debug("closed tokenMan")
-					return
-				}
-				log.Debug("force token refresh")
-			case <-time.After(t.token.refreshIn()):
-				log.Debug("auto refresh token")
-			}
+	for {
+		select {
+		case <-t.closed:
+			return
+		case <-t.refreshTimer:
+			log.Debug("auto refresh token")
+			t.retrieveNewToken()
+			t.refreshTimer = time.After(t.token.refreshIn())
+		case <-t.getTokenChan:
+			token := t.token
+			t.returnTokenChan <- token
+		case <-t.invalidateTokenChan:
+			t.retrieveNewToken()
+			t.refreshTimer = time.After(t.token.refreshIn())
+			t.invalidateDone <- true
 		}
-	}()
-}
-
-func (t *tokenMan) invalidateToken() {
-	log.Debug("invalidating token")
-	t.token = nil
-	t.doRefresh <- true
+	}
 }
 
 // will block until valid
-func (t *tokenMan) getToken() *oauthToken {
-	getTokenLock.Lock()
-	defer getTokenLock.Unlock()
-
-	if t.token.isValid() {
-		log.Debugf("returning existing token: %v", t.token)
-		return t.token
+func (t *tokenMan) invalidateToken() {
+	//has been closed
+	if atomic.LoadInt32(t.isClosed) == int32(1) {
+		log.Debug("TokenManager: invalidateToken() called on closed tokenManager")
+		return
 	}
-
-	t.retrieveNewToken()
-	return t.token
+	log.Debug("invalidating token")
+	t.invalidateTokenChan <- true
+	<-t.invalidateDone
 }
 
+func (t *tokenMan) getToken() *oauthToken {
+	//has been closed
+	if atomic.LoadInt32(t.isClosed) == int32(1) {
+		log.Debug("TokenManager: getToken() called on closed tokenManager")
+		return nil
+	}
+	t.getTokenChan <- true
+	return <-t.returnTokenChan
+}
+
+/*
+ * blocking close() of tokenMan
+ */
+
 func (t *tokenMan) close() {
+	//has been closed
+	if atomic.SwapInt32(t.isClosed, 1) == int32(1) {
+		log.Panic("TokenManager: close() has been called before!")
+		return
+	}
 	log.Debug("close token manager")
-	close(t.doRefresh)
+	t.quitPollingForToken <- true
+	// sending instead of closing, to make sure it enters the t.doRefresh branch
+	t.closed <- true
+	close(t.closed)
+	log.Debug("token manager closed")
 }
 
 // don't call externally. will block until success.
@@ -99,18 +130,11 @@
 	}
 	uri.Path = path.Join(uri.Path, "/accesstoken")
 
-	retryIn := 5 * time.Millisecond
-	maxBackOff := maxBackoffTimeout
-	backOffFunc := createBackOff(retryIn, maxBackOff)
-	first := true
+	pollWithBackoff(t.quitPollingForToken, t.getRetrieveNewTokenClosure(uri), func(err error) { log.Errorf("Error getting new token : ", err) })
+}
 
-	for {
-		if first {
-			first = false
-		} else {
-			backOffFunc()
-		}
-
+func (t *tokenMan) getRetrieveNewTokenClosure(uri *url.URL) func(chan bool) error {
+	return func(_ chan bool) error {
 		form := url.Values{}
 		form.Set("grant_type", "client_credentials")
 		form.Add("client_id", config.GetString(configConsumerKey))
@@ -133,26 +157,26 @@
 		resp, err := client.Do(req)
 		if err != nil {
 			log.Errorf("Unable to Connect to Edge Proxy Server: %v", err)
-			continue
+			return err
 		}
 
 		body, err := ioutil.ReadAll(resp.Body)
 		resp.Body.Close()
 		if err != nil {
 			log.Errorf("Unable to read EdgeProxy Sever response: %v", err)
-			continue
+			return err
 		}
 
 		if resp.StatusCode != 200 {
 			log.Errorf("Oauth Request Failed with Resp Code: %d. Body: %s", resp.StatusCode, string(body))
-			continue
+			return expected200Error{}
 		}
 
 		var token oauthToken
 		err = json.Unmarshal(body, &token)
 		if err != nil {
 			log.Errorf("unable to unmarshal JSON response '%s': %v", string(body), err)
-			continue
+			return err
 		}
 
 		if token.ExpiresIn > 0 {
@@ -166,12 +190,17 @@
 
 		if newInstanceID {
 			newInstanceID = false
-			updateApidInstanceInfo()
-		}
+			err = updateApidInstanceInfo()
+			if err != nil {
+				log.Errorf("unable to unmarshal update apid instance info : %v", string(body), err)
+				return err
 
+			}
+		}
 		t.token = &token
 		config.Set(configBearerToken, token.AccessToken)
-		return
+
+		return nil
 	}
 }
 
diff --git a/token_test.go b/token_test.go
index 5deec1c..045b318 100644
--- a/token_test.go
+++ b/token_test.go
@@ -1,5 +1,8 @@
 package apidApigeeSync
 
+/*
+ * Unit test of token manager
+ */
 import (
 	"time"
 
@@ -17,6 +20,8 @@
 	Context("oauthToken", func() {
 
 		It("should calculate valid token", func() {
+			log.Info("Starting token tests...")
+
 			t := &oauthToken{
 				AccessToken: "x",
 				ExpiresIn:   120000,
@@ -25,9 +30,10 @@
 			Expect(t.refreshIn().Seconds()).To(BeNumerically(">", 0))
 			Expect(t.needsRefresh()).To(BeFalse())
 			Expect(t.isValid()).To(BeTrue())
-		})
+		}, 3)
 
 		It("should calculate expired token", func() {
+
 			t := &oauthToken{
 				AccessToken: "x",
 				ExpiresIn:   0,
@@ -36,9 +42,10 @@
 			Expect(t.refreshIn().Seconds()).To(BeNumerically("<", 0))
 			Expect(t.needsRefresh()).To(BeTrue())
 			Expect(t.isValid()).To(BeFalse())
-		})
+		}, 3)
 
 		It("should calculate token needing refresh", func() {
+
 			t := &oauthToken{
 				AccessToken: "x",
 				ExpiresIn:   59000,
@@ -47,44 +54,76 @@
 			Expect(t.refreshIn().Seconds()).To(BeNumerically("<", 0))
 			Expect(t.needsRefresh()).To(BeTrue())
 			Expect(t.isValid()).To(BeTrue())
-		})
+		}, 3)
 
 		It("should calculate on empty token", func() {
+
 			t := &oauthToken{}
 			Expect(t.refreshIn().Seconds()).To(BeNumerically("<=", 0))
 			Expect(t.needsRefresh()).To(BeTrue())
 			Expect(t.isValid()).To(BeFalse())
-		})
+		}, 3)
 	})
 
 	Context("tokenMan", func() {
 
 		It("should get a valid token", func() {
-			token := tokenManager.getToken()
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				defer GinkgoRecover()
+
+				res := oauthToken{
+					AccessToken: "ABCD",
+					ExpiresIn:   1000,
+				}
+				body, err := json.Marshal(res)
+				Expect(err).NotTo(HaveOccurred())
+				w.Write(body)
+			}))
+			config.Set(configProxyServerBaseURI, ts.URL)
+			testedTokenManager := createTokenManager()
+			token := testedTokenManager.getToken()
 
 			Expect(token.AccessToken).ToNot(BeEmpty())
 			Expect(token.ExpiresIn > 0).To(BeTrue())
 			Expect(token.ExpiresAt).To(BeTemporally(">", time.Now()))
 
-			bToken := tokenManager.getBearerToken()
+			bToken := testedTokenManager.getBearerToken()
 			Expect(bToken).To(Equal(token.AccessToken))
-		})
+			testedTokenManager.close()
+			ts.Close()
+		}, 3)
 
 		It("should refresh when forced to", func() {
-			token := tokenManager.getToken()
+
+			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				defer GinkgoRecover()
+
+				res := oauthToken{
+					AccessToken: generateUUID(),
+					ExpiresIn:   1000,
+				}
+				body, err := json.Marshal(res)
+				Expect(err).NotTo(HaveOccurred())
+				w.Write(body)
+			}))
+			config.Set(configProxyServerBaseURI, ts.URL)
+
+			testedTokenManager := createTokenManager()
+			token := testedTokenManager.getToken()
 			Expect(token.AccessToken).ToNot(BeEmpty())
 
-			tokenManager.invalidateToken()
+			testedTokenManager.invalidateToken()
 
-			token2 := tokenManager.getToken()
+			token2 := testedTokenManager.getToken()
 			Expect(token).ToNot(Equal(token2))
 			Expect(token.AccessToken).ToNot(Equal(token2.AccessToken))
-		})
+			testedTokenManager.close()
+			ts.Close()
+		}, 3)
 
 		It("should refresh in refresh interval", func(done Done) {
 
-			finished := make(chan bool)
-			var tm *tokenMan
+			finished := make(chan bool, 1)
 			start := time.Now()
 			count := 0
 			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -106,33 +145,27 @@
 				Expect(err).NotTo(HaveOccurred())
 				w.Write(body)
 			}))
-			defer ts.Close()
 
-			tokenManager.getToken()
-			tokenManager.close()
-			oldBase := config.Get(configProxyServerBaseURI)
 			config.Set(configProxyServerBaseURI, ts.URL)
-			oldFloat := refreshFloatTime
-			refreshFloatTime = 950 * time.Millisecond
-			defer func() {
-				tm.close()
-				config.Set(configProxyServerBaseURI, oldBase)
-				tokenManager = createTokenManager()
-				refreshFloatTime = oldFloat
-			}()
+			testedTokenManager := createTokenManager()
 
-			tm = createTokenManager()
+			testedTokenManager.getToken()
+
 			<-finished
+
+			testedTokenManager.close()
+			ts.Close()
+
 			close(done)
-		})
+		}, 3)
 
 		It("should have created_at_apid first time, update_at_apid after", func(done Done) {
-
-			finished := make(chan bool)
-			var tm *tokenMan
+			finished := make(chan bool, 1)
 			count := 0
+
+			newInstanceID = true
+
 			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-				defer GinkgoRecover()
 
 				count++
 				if count == 1 {
@@ -147,29 +180,23 @@
 				}
 				res := oauthToken{
 					AccessToken: string(count),
-					ExpiresIn:   2000,
+					ExpiresIn:   200000,
 				}
 				body, err := json.Marshal(res)
 				Expect(err).NotTo(HaveOccurred())
 				w.Write(body)
 			}))
-			defer ts.Close()
 
-			tokenManager.getToken()
-			tokenManager.close()
-			oldBase := config.Get(configProxyServerBaseURI)
 			config.Set(configProxyServerBaseURI, ts.URL)
-			defer func() {
-				tm.close()
-				config.Set(configProxyServerBaseURI, oldBase)
-				tokenManager = createTokenManager()
-			}()
+			testedTokenManager := createTokenManager()
 
-			newInstanceID = true
-			tm = createTokenManager()
-			tm.invalidateToken()
+			testedTokenManager.getToken()
+			testedTokenManager.invalidateToken()
+			testedTokenManager.getToken()
 			<-finished
+			testedTokenManager.close()
+			ts.Close()
 			close(done)
-		})
+		}, 3)
 	})
 })