diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5737e823e9..283d33f0f7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,25 @@
 
 ## HEAD
 
+### CTFE Storage Saving: Extra Data Issuance Chain Deduplication
+
+To reduce CT/Trillian database storage by deduplication of the entire issuance chain (intermediate certificate(s) and root certificate) that is currently stored in the Trillian merkle tree leaf ExtraData field. Storage cost should be reduced by at least 33% for new CT logs with this feature enabled. Currently only MySQL/MariaDB is supported to store the issuance chain in the CTFE database.
+
+Existing logs are not affected by this change. 
+
+Log operators can choose to opt-in this change for new CT logs by adding new CTFE configs in the [LogMultiConfig](trillian/ctfe/configpb/config.proto) and importing the [database schema](trillian/ctfe/storage/mysql/schema.sql). See [example](trillian/examples/deployment/docker/ctfe/ct_server.cfg).
+
+- `ctfe_storage_connection_string`
+- `extra_data_issuance_chain_storage_backend`
+
+An optional LRU cache can be enabled by providing the following flags.
+
+- `cache_type`
+- `cache_size`
+- `cache_ttl`
+
+This change is tested in Cloud Build tests using the `mysql:8.4` Docker image as of the time of writing.
+
 ### Submission proxy: Root compatibility checking
 
 * Adds the ability for a CT client to disable root compatibile checking: https://github.com/google/certificate-transparency-go/pull/1258
diff --git a/cloudbuild.yaml b/cloudbuild.yaml
index 6c856c11c1..50ac5ef7b1 100644
--- a/cloudbuild.yaml
+++ b/cloudbuild.yaml
@@ -85,6 +85,13 @@ steps:
 
       # Wait for trillian logserver to be up
       until nc -z deployment_trillian-log-server_1 8090; do echo .; sleep 5; done
+
+      # Reset the CT test database
+      export CT_GO_PATH="$$(go list -f '{{.Dir}}' github.com/google/certificate-transparency-go)"
+      export MYSQL_HOST="mysql"
+      export MYSQL_ROOT_PASSWORD="zaphod"
+      export MYSQL_USER_HOST="%"
+      yes | bash "$${CT_GO_PATH}/scripts/resetctdb.sh" --verbose
   waitFor: ['prepare']
 
 # Run the presubmit tests
diff --git a/cloudbuild_master.yaml b/cloudbuild_master.yaml
index 0e47e326f6..566edfe0f9 100644
--- a/cloudbuild_master.yaml
+++ b/cloudbuild_master.yaml
@@ -85,6 +85,13 @@ steps:
 
       # Wait for trillian logserver to be up
       until nc -z deployment_trillian-log-server_1 8090; do echo .; sleep 5; done
+
+      # Reset the CT test database
+      export CT_GO_PATH="$$(go list -f '{{.Dir}}' github.com/google/certificate-transparency-go)"
+      export MYSQL_HOST="mysql"
+      export MYSQL_ROOT_PASSWORD="zaphod"
+      export MYSQL_USER_HOST="%"
+      yes | bash "$${CT_GO_PATH}/scripts/resetctdb.sh" --verbose
   waitFor: ['prepare']
 
 # Run the presubmit tests
diff --git a/integration/Dockerfile b/integration/Dockerfile
index 8509fe1074..96b59f5e8b 100644
--- a/integration/Dockerfile
+++ b/integration/Dockerfile
@@ -7,7 +7,7 @@ WORKDIR /testbase
 ARG GOFLAGS=""
 ENV GOFLAGS=$GOFLAGS
 
-RUN apt-get update && apt-get -y install build-essential curl docker-compose lsof netcat-openbsd unzip wget xxd
+RUN apt-get update && apt-get -y install build-essential curl docker-compose lsof mariadb-client netcat-openbsd unzip wget xxd
 
 RUN cd /usr/bin && curl -L -O https://github.com/jqlang/jq/releases/download/jq-1.7/jq-linux64 && mv jq-linux64 /usr/bin/jq && chmod +x /usr/bin/jq
 RUN curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.55.1
diff --git a/scripts/resetctdb.sh b/scripts/resetctdb.sh
index 438bbd8695..488f2b7630 100755
--- a/scripts/resetctdb.sh
+++ b/scripts/resetctdb.sh
@@ -41,6 +41,7 @@ collect_vars() {
   # handle flags
   FORCE=false
   VERBOSE=false
+
   while [[ $# -gt 0 ]]; do
     case "$1" in
       --force) FORCE=true ;;
@@ -74,16 +75,21 @@ main() {
 
   if [ -z ${REPLY+x} ] || [[ $REPLY =~ ^[Yy]$ ]]
   then
-      echo "Resetting DB..."
-      mysql "${FLAGS[@]}" -e "DROP DATABASE IF EXISTS ${MYSQL_DATABASE};" || \
-        die "Error: Failed to drop database '${MYSQL_DATABASE}'."
-      mysql "${FLAGS[@]}" -e "CREATE DATABASE ${MYSQL_DATABASE};" || \
-        die "Error: Failed to create database '${MYSQL_DATABASE}'."
-      mysql "${FLAGS[@]}" -e "CREATE USER IF NOT EXISTS ${MYSQL_USER}@'${MYSQL_USER_HOST}' IDENTIFIED BY '${MYSQL_PASSWORD}';" || \
-        die "Error: Failed to create user '${MYSQL_USER}@${MYSQL_USER_HOST}'."
-      mysql "${FLAGS[@]}" -e "GRANT ALL ON ${MYSQL_DATABASE}.* TO ${MYSQL_USER}@'${MYSQL_USER_HOST}'" || \
-        die "Error: Failed to grant '${MYSQL_USER}' user all privileges on '${MYSQL_DATABASE}'."
-      echo "Reset Complete"
+    echo "Resetting DB..."
+    mysql "${FLAGS[@]}" -e "DROP DATABASE IF EXISTS ${MYSQL_DATABASE};" || \
+      die "Error: Failed to drop database '${MYSQL_DATABASE}'."
+    mysql "${FLAGS[@]}" -e "CREATE DATABASE ${MYSQL_DATABASE};" || \
+      die "Error: Failed to create database '${MYSQL_DATABASE}'."
+    mysql "${FLAGS[@]}" -e "CREATE USER IF NOT EXISTS ${MYSQL_USER}@'${MYSQL_USER_HOST}' IDENTIFIED BY '${MYSQL_PASSWORD}';" || \
+      die "Error: Failed to create user '${MYSQL_USER}@${MYSQL_USER_HOST}'."
+    mysql "${FLAGS[@]}" -e "GRANT ALL ON ${MYSQL_DATABASE}.* TO ${MYSQL_USER}@'${MYSQL_USER_HOST}';" || \
+      die "Error: Failed to grant '${MYSQL_USER}' user all privileges on '${MYSQL_DATABASE}'."
+    mysql "${FLAGS[@]}" -e "FLUSH PRIVILEGES;" || \
+      die "Error: Failed to flush privileges."
+    mysql "${FLAGS[@]}" -D ${MYSQL_DATABASE} < ${CT_GO_PATH}/trillian/ctfe/storage/mysql/schema.sql || \
+      die "Error: Failed to import schema in '${MYSQL_DATABASE}' database."
+
+    echo "Reset Complete"
   fi
 }
 
diff --git a/trillian/ctfe/cache/cache.go b/trillian/ctfe/cache/cache.go
index 3ef06cb970..edd6b8059a 100644
--- a/trillian/ctfe/cache/cache.go
+++ b/trillian/ctfe/cache/cache.go
@@ -15,7 +15,30 @@
 // Package cache defines the IssuanceChainCache type, which allows different cache implementation with Get and Set operations.
 package cache
 
-import "context"
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/google/certificate-transparency-go/trillian/ctfe/cache/lru"
+	"github.com/google/certificate-transparency-go/trillian/ctfe/cache/noop"
+)
+
+// Type represents the cache type.
+type Type string
+
+// Type constants for the cache type.
+const (
+	Unknown Type = ""
+	NOOP    Type = "noop"
+	LRU     Type = "lru"
+)
+
+// Option represents the cache option, which includes the cache size and time-to-live.
+type Option struct {
+	Size int
+	TTL  time.Duration
+}
 
 // IssuanceChainCache is an interface which allows CTFE binaries to use different cache implementations for issuance chains.
 type IssuanceChainCache interface {
@@ -25,3 +48,21 @@ type IssuanceChainCache interface {
 	// Set inserts the key-value pair of issuance chain.
 	Set(ctx context.Context, key []byte, chain []byte) error
 }
+
+// NewIssuanceChainCache returns noop.IssuanceChainCache for noop type or lru.IssuanceChainCache for lru cache type.
+func NewIssuanceChainCache(_ context.Context, cacheType Type, option Option) (IssuanceChainCache, error) {
+	switch cacheType {
+	case Unknown, NOOP:
+		return &noop.IssuanceChainCache{}, nil
+	case LRU:
+		if option.Size < 0 {
+			return nil, errors.New("invalid cache_size flag")
+		}
+		if option.TTL < 0*time.Second {
+			return nil, errors.New("invalid cache_ttl flag")
+		}
+		return lru.NewIssuanceChainCache(lru.CacheOption{Size: option.Size, TTL: option.TTL}), nil
+	}
+
+	return nil, errors.New("invalid cache_type flag")
+}
diff --git a/trillian/ctfe/cache/noop/noop.go b/trillian/ctfe/cache/noop/noop.go
new file mode 100644
index 0000000000..ed8f4ed9bf
--- /dev/null
+++ b/trillian/ctfe/cache/noop/noop.go
@@ -0,0 +1,28 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package noop defines the IssuanceChainCache type, which implements IssuanceChainCache interface with Get and Set operations.
+package noop
+
+import "context"
+
+type IssuanceChainCache struct{}
+
+func (c *IssuanceChainCache) Get(_ context.Context, key []byte) ([]byte, error) {
+	return nil, nil
+}
+
+func (c *IssuanceChainCache) Set(_ context.Context, key []byte, chain []byte) error {
+	return nil
+}
diff --git a/trillian/ctfe/ct_server/main.go b/trillian/ctfe/ct_server/main.go
index 0aa27a21df..4e20f58b43 100644
--- a/trillian/ctfe/ct_server/main.go
+++ b/trillian/ctfe/ct_server/main.go
@@ -32,6 +32,7 @@ import (
 	"time"
 
 	"github.com/google/certificate-transparency-go/trillian/ctfe"
+	"github.com/google/certificate-transparency-go/trillian/ctfe/cache"
 	"github.com/google/certificate-transparency-go/trillian/ctfe/configpb"
 	"github.com/google/trillian"
 	"github.com/google/trillian/crypto/keys"
@@ -73,6 +74,9 @@ var (
 	quotaIntermediate  = flag.Bool("quota_intermediate", true, "Enable requesting of quota for intermediate certificates in submitted chains")
 	handlerPrefix      = flag.String("handler_prefix", "", "If set e.g. to '/logs' will prefix all handlers that don't define a custom prefix")
 	pkcs11ModulePath   = flag.String("pkcs11_module_path", "", "Path to the PKCS#11 module to use for keys that use the PKCS#11 interface")
+	cacheType          = flag.String("cache_type", "noop", "Supported cache type: noop, lru (Default: noop)")
+	cacheSize          = flag.Int("cache_size", -1, "Size parameter set to 0 makes cache of unlimited size")
+	cacheTTL           = flag.Duration("cache_ttl", -1*time.Second, "Providing 0 TTL turns expiring off")
 )
 
 const unknownRemoteUser = "UNKNOWN_REMOTE"
@@ -218,7 +222,19 @@ func main() {
 	// client.
 	var publicKeys []crypto.PublicKey
 	for _, c := range cfg.LogConfigs.Config {
-		inst, err := setupAndRegister(ctx, clientMap[c.LogBackendName], *rpcDeadline, c, corsMux, *handlerPrefix, *maskInternalErrors)
+		inst, err := setupAndRegister(ctx,
+			clientMap[c.LogBackendName],
+			*rpcDeadline,
+			c,
+			corsMux,
+			*handlerPrefix,
+			*maskInternalErrors,
+			cache.Type(*cacheType),
+			cache.Option{
+				Size: *cacheSize,
+				TTL:  *cacheTTL,
+			},
+		)
 		if err != nil {
 			klog.Exitf("Failed to set up log instance for %+v: %v", cfg, err)
 		}
@@ -330,7 +346,7 @@ func awaitSignal(doneFn func()) {
 	doneFn()
 }
 
-func setupAndRegister(ctx context.Context, client trillian.TrillianLogClient, deadline time.Duration, cfg *configpb.LogConfig, mux *http.ServeMux, globalHandlerPrefix string, maskInternalErrors bool) (*ctfe.Instance, error) {
+func setupAndRegister(ctx context.Context, client trillian.TrillianLogClient, deadline time.Duration, cfg *configpb.LogConfig, mux *http.ServeMux, globalHandlerPrefix string, maskInternalErrors bool, cacheType cache.Type, cacheOption cache.Option) (*ctfe.Instance, error) {
 	vCfg, err := ctfe.ValidateLogConfig(cfg)
 	if err != nil {
 		return nil, err
@@ -343,6 +359,8 @@ func setupAndRegister(ctx context.Context, client trillian.TrillianLogClient, de
 		MetricFactory:      prometheus.MetricFactory{},
 		RequestLog:         new(ctfe.DefaultRequestLog),
 		MaskInternalErrors: maskInternalErrors,
+		CacheType:          cacheType,
+		CacheOption:        cacheOption,
 	}
 	if *quotaRemote {
 		klog.Info("Enabling quota for requesting IP")
diff --git a/trillian/ctfe/handlers.go b/trillian/ctfe/handlers.go
index 303fcae375..2ade820d3d 100644
--- a/trillian/ctfe/handlers.go
+++ b/trillian/ctfe/handlers.go
@@ -278,6 +278,8 @@ type logInfo struct {
 	signer crypto.Signer
 	// sthGetter provides STHs for the log
 	sthGetter STHGetter
+	// issuanceChainService provides the issuance chain add and get operations
+	issuanceChainService *issuanceChainService
 }
 
 // newLogInfo creates a new instance of logInfo.
@@ -286,6 +288,7 @@ func newLogInfo(
 	validationOpts CertValidationOpts,
 	signer crypto.Signer,
 	timeSource util.TimeSource,
+	issuanceChainService *issuanceChainService,
 ) *logInfo {
 	vCfg := instanceOpts.Validated
 	cfg := vCfg.Config
@@ -330,6 +333,8 @@ func newLogInfo(
 	maxMergeDelay.Set(float64(cfg.MaxMergeDelaySec), label)
 	expMergeDelay.Set(float64(cfg.ExpectedMergeDelaySec), label)
 
+	li.issuanceChainService = issuanceChainService
+
 	return li
 }
 
@@ -465,15 +470,15 @@ func addChainInternal(ctx context.Context, li *logInfo, w http.ResponseWriter, r
 	if err != nil {
 		return http.StatusBadRequest, fmt.Errorf("failed to build MerkleTreeLeaf: %s", err)
 	}
-	leaf, err := buildLogLeafForAddChain(li, *merkleLeaf, chain, isPrecert)
+	leaf, err := li.issuanceChainService.BuildLogLeaf(ctx, chain, li.LogPrefix, merkleLeaf, isPrecert)
 	if err != nil {
-		return http.StatusInternalServerError, fmt.Errorf("failed to build LogLeaf: %s", err)
+		return http.StatusInternalServerError, err
 	}
 
 	// Send the Merkle tree leaf on to the Log server.
 	req := trillian.QueueLeafRequest{
 		LogId:    li.logID,
-		Leaf:     &leaf,
+		Leaf:     leaf,
 		ChargeTo: li.chargeUser(r),
 	}
 	if li.instanceOpts.CertificateQuotaUser != nil {
@@ -745,10 +750,11 @@ func getEntries(ctx context.Context, li *logInfo, w http.ResponseWriter, r *http
 		Count:      count,
 		ChargeTo:   li.chargeUser(r),
 	}
-	rsp, err := li.rpcClient.GetLeavesByRange(ctx, &req)
+	rsp, httpStatus, err := rpcGetLeavesByRange(ctx, li, &req)
 	if err != nil {
-		return li.toHTTPStatus(err), fmt.Errorf("backend GetLeavesByRange request failed: %s", err)
+		return httpStatus, err
 	}
+
 	var currentRoot types.LogRootV1
 	if err := currentRoot.UnmarshalBinary(rsp.GetSignedLogRoot().GetLogRoot()); err != nil {
 		return http.StatusInternalServerError, fmt.Errorf("failed to unmarshal root: %v", rsp.GetSignedLogRoot().GetLogRoot())
@@ -798,6 +804,21 @@ func getEntries(ctx context.Context, li *logInfo, w http.ResponseWriter, r *http
 	return http.StatusOK, nil
 }
 
+// rpcGetLeavesByRange calls Trillian GetLeavesByRange RPC and fixes issuance chain in each log leaf if necessary.
+func rpcGetLeavesByRange(ctx context.Context, li *logInfo, req *trillian.GetLeavesByRangeRequest) (*trillian.GetLeavesByRangeResponse, int, error) {
+	rsp, err := li.rpcClient.GetLeavesByRange(ctx, req)
+	if err != nil {
+		return nil, li.toHTTPStatus(err), fmt.Errorf("backend GetLeavesByRange request failed: %s", err)
+	}
+	for _, leaf := range rsp.Leaves {
+		if err := li.issuanceChainService.FixLogLeaf(ctx, leaf); err != nil {
+			return nil, http.StatusInternalServerError, fmt.Errorf("failed to fix log leaf: %v", rsp)
+		}
+	}
+
+	return rsp, http.StatusOK, nil
+}
+
 func getRoots(_ context.Context, li *logInfo, w http.ResponseWriter, _ *http.Request) (int, error) {
 	// Pull out the raw certificates from the parsed versions
 	rawCerts := make([][]byte, 0, len(li.validationOpts.trustedRoots.RawCertificates()))
@@ -834,9 +855,9 @@ func getEntryAndProof(ctx context.Context, li *logInfo, w http.ResponseWriter, r
 		TreeSize:  treeSize,
 		ChargeTo:  li.chargeUser(r),
 	}
-	rsp, err := li.rpcClient.GetEntryAndProof(ctx, &req)
+	rsp, httpStatus, err := rpcGetEntryAndProof(ctx, li, &req)
 	if err != nil {
-		return li.toHTTPStatus(err), fmt.Errorf("backend GetEntryAndProof request failed: %s", err)
+		return httpStatus, err
 	}
 
 	var currentRoot types.LogRootV1
@@ -881,6 +902,19 @@ func getEntryAndProof(ctx context.Context, li *logInfo, w http.ResponseWriter, r
 	return http.StatusOK, nil
 }
 
+// rpcGetEntryAndProof calls Trillian GetEntryAndProof RPC and fixes issuance chain in the log leaf if necessary.
+func rpcGetEntryAndProof(ctx context.Context, li *logInfo, req *trillian.GetEntryAndProofRequest) (*trillian.GetEntryAndProofResponse, int, error) {
+	rsp, err := li.rpcClient.GetEntryAndProof(ctx, req)
+	if err != nil {
+		return nil, li.toHTTPStatus(err), fmt.Errorf("backend GetEntryAndProof request failed: %s", err)
+	}
+	if err := li.issuanceChainService.FixLogLeaf(ctx, rsp.Leaf); err != nil {
+		return nil, http.StatusInternalServerError, fmt.Errorf("failed to fix log leaf: %v", rsp)
+	}
+
+	return rsp, http.StatusOK, nil
+}
+
 // getRPCDeadlineTime calculates the future time an RPC should expire based on our config
 func getRPCDeadlineTime(li *logInfo) time.Time {
 	return li.TimeSource.Now().Add(li.instanceOpts.Deadline)
@@ -923,15 +957,6 @@ func extractRawCerts(chain []*x509.Certificate) []ct.ASN1Cert {
 	return raw
 }
 
-// buildLogLeafForAddChain does the hashing to build a LogLeaf that will be
-// sent to the backend by add-chain and add-pre-chain endpoints.
-func buildLogLeafForAddChain(li *logInfo,
-	merkleLeaf ct.MerkleTreeLeaf, chain []*x509.Certificate, isPrecert bool,
-) (trillian.LogLeaf, error) {
-	raw := extractRawCerts(chain)
-	return util.BuildLogLeaf(li.LogPrefix, merkleLeaf, 0, raw[0], raw[1:], isPrecert)
-}
-
 // marshalAndWriteAddChainResponse is used by add-chain and add-pre-chain to create and write
 // the JSON response to the client
 func marshalAndWriteAddChainResponse(sct *ct.SignedCertificateTimestamp, signer crypto.Signer, w http.ResponseWriter) error {
diff --git a/trillian/ctfe/handlers_test.go b/trillian/ctfe/handlers_test.go
index bc786409a0..ada5220cbd 100644
--- a/trillian/ctfe/handlers_test.go
+++ b/trillian/ctfe/handlers_test.go
@@ -165,7 +165,7 @@ func setupTest(t *testing.T, pemRoots []string, signer crypto.Signer) handlerTes
 	cfg := &configpb.LogConfig{LogId: 0x42, Prefix: "test", IsMirror: false}
 	vCfg := &ValidatedLogConfig{Config: cfg}
 	iOpts := InstanceOptions{Validated: vCfg, Client: info.client, Deadline: time.Millisecond * 500, MetricFactory: monitoring.InertMetricFactory{}, RequestLog: new(DefaultRequestLog)}
-	info.li = newLogInfo(iOpts, vOpts, signer, fakeTimeSource)
+	info.li = newLogInfo(iOpts, vOpts, signer, fakeTimeSource, newIssuanceChainService(nil, nil))
 
 	for _, pemRoot := range pemRoots {
 		if !info.roots.AppendCertsFromPEM([]byte(pemRoot)) {
diff --git a/trillian/ctfe/instance.go b/trillian/ctfe/instance.go
index 0056fd0756..6fe480ce03 100644
--- a/trillian/ctfe/instance.go
+++ b/trillian/ctfe/instance.go
@@ -29,6 +29,8 @@ import (
 
 	"github.com/google/certificate-transparency-go/asn1"
 	"github.com/google/certificate-transparency-go/schedule"
+	"github.com/google/certificate-transparency-go/trillian/ctfe/cache"
+	"github.com/google/certificate-transparency-go/trillian/ctfe/storage"
 	"github.com/google/certificate-transparency-go/trillian/util"
 	"github.com/google/certificate-transparency-go/x509"
 	"github.com/google/certificate-transparency-go/x509util"
@@ -72,6 +74,10 @@ type InstanceOptions struct {
 	// MaskInternalErrors indicates if internal server errors should be masked
 	// or returned to the user containing the full error message.
 	MaskInternalErrors bool
+	// CacheType is the CTFE cache type.
+	CacheType cache.Type
+	// CacheOption includes the cache size and time-to-live (TTL).
+	CacheOption cache.Option
 }
 
 // Instance is a set up log/mirror instance. It must be created with the
@@ -174,7 +180,20 @@ func setUpLogInfo(ctx context.Context, opts InstanceOptions) (*logInfo, error) {
 		return nil, fmt.Errorf("failed to parse RejectExtensions: %v", err)
 	}
 
-	logInfo := newLogInfo(opts, validationOpts, signer, new(util.SystemTimeSource))
+	// Initialise IssuanceChainService with IssuanceChainStorage and IssuanceChainCache.
+	issuanceChainStorage, err := storage.NewIssuanceChainStorage(ctx, vCfg.ExtraDataIssuanceChainStorageBackend, vCfg.CTFEStorageConnectionString)
+	if err != nil {
+		return nil, err
+	}
+	// issuanceChainCache is nil if the cache related flags are not defined.
+	issuanceChainCache, err := cache.NewIssuanceChainCache(ctx, opts.CacheType, opts.CacheOption)
+	if err != nil {
+		return nil, err
+	}
+
+	issuanceChainService := newIssuanceChainService(issuanceChainStorage, issuanceChainCache)
+
+	logInfo := newLogInfo(opts, validationOpts, signer, new(util.SystemTimeSource), issuanceChainService)
 	return logInfo, nil
 }
 
diff --git a/trillian/ctfe/instance_test.go b/trillian/ctfe/instance_test.go
index 10cf9e27b5..c191a5157a 100644
--- a/trillian/ctfe/instance_test.go
+++ b/trillian/ctfe/instance_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	ct "github.com/google/certificate-transparency-go"
+	"github.com/google/certificate-transparency-go/trillian/ctfe/cache"
 	"github.com/google/certificate-transparency-go/trillian/ctfe/configpb"
 	"github.com/google/trillian/crypto/keys"
 	"github.com/google/trillian/crypto/keys/pem"
@@ -257,7 +258,7 @@ func TestSetUpInstanceSetsValidationOpts(t *testing.T) {
 			if err != nil {
 				t.Fatalf("ValidateLogConfig(): %v", err)
 			}
-			opts := InstanceOptions{Validated: vCfg, Deadline: time.Second, MetricFactory: monitoring.InertMetricFactory{}}
+			opts := InstanceOptions{Validated: vCfg, Deadline: time.Second, MetricFactory: monitoring.InertMetricFactory{}, CacheType: cache.NOOP, CacheOption: cache.Option{}}
 
 			inst, err := SetUpInstance(ctx, opts)
 			if err != nil {
diff --git a/trillian/ctfe/services.go b/trillian/ctfe/services.go
index 580cf8763e..46c3e1cba4 100644
--- a/trillian/ctfe/services.go
+++ b/trillian/ctfe/services.go
@@ -17,10 +17,19 @@ package ctfe
 import (
 	"context"
 	"crypto/sha256"
+	"errors"
+	"fmt"
 
+	"github.com/google/certificate-transparency-go/asn1"
+	"github.com/google/certificate-transparency-go/tls"
 	"github.com/google/certificate-transparency-go/trillian/ctfe/cache"
 	"github.com/google/certificate-transparency-go/trillian/ctfe/storage"
+	"github.com/google/certificate-transparency-go/trillian/util"
+	"github.com/google/certificate-transparency-go/x509"
+	"github.com/google/trillian"
 	"k8s.io/klog/v2"
+
+	ct "github.com/google/certificate-transparency-go"
 )
 
 type issuanceChainService struct {
@@ -37,8 +46,17 @@ func newIssuanceChainService(s storage.IssuanceChainStorage, c cache.IssuanceCha
 	return service
 }
 
+func (s *issuanceChainService) isCTFEStorageEnabled() bool {
+	return s.storage != nil
+}
+
 // GetByHash returns the issuance chain with hash as the input.
 func (s *issuanceChainService) GetByHash(ctx context.Context, hash []byte) ([]byte, error) {
+	// Return err if CTFE storage backend is not enabled.
+	if !s.isCTFEStorageEnabled() {
+		return nil, errors.New("failed to GetByHash when storage is nil")
+	}
+
 	// Return if found in cache.
 	chain, err := s.cache.Get(ctx, hash)
 	if chain != nil || err != nil {
@@ -62,9 +80,14 @@ func (s *issuanceChainService) GetByHash(ctx context.Context, hash []byte) ([]by
 	return chain, nil
 }
 
-// Add adds the issuance chain into the storage and cache and returns the hash
+// add adds the issuance chain into the storage and cache and returns the hash
 // of the chain.
-func (s *issuanceChainService) Add(ctx context.Context, chain []byte) ([]byte, error) {
+func (s *issuanceChainService) add(ctx context.Context, chain []byte) ([]byte, error) {
+	// Return err if CTFE storage backend is not enabled.
+	if !s.isCTFEStorageEnabled() {
+		return nil, errors.New("failed to Add when storage is nil")
+	}
+
 	hash := issuanceChainHash(chain)
 
 	if err := s.storage.Add(ctx, hash, chain); err != nil {
@@ -82,6 +105,118 @@ func (s *issuanceChainService) Add(ctx context.Context, chain []byte) ([]byte, e
 	return hash, nil
 }
 
+// BuildLogLeaf builds the MerkleTreeLeaf that gets sent to the backend, and make a trillian.LogLeaf for it.
+func (s *issuanceChainService) BuildLogLeaf(ctx context.Context, chain []*x509.Certificate, logPrefix string, merkleLeaf *ct.MerkleTreeLeaf, isPrecert bool) (*trillian.LogLeaf, error) {
+	raw := extractRawCerts(chain)
+
+	// If CTFE storage is enabled for issuance chain, add the chain to storage
+	// and cache, and then build log leaf. If Trillian gRPC is enabled for
+	// issuance chain, build the log leaf.
+	if s.isCTFEStorageEnabled() {
+		issuanceChain, err := asn1.Marshal(raw[1:])
+		if err != nil {
+			return &trillian.LogLeaf{}, fmt.Errorf("failed to marshal issuance chain: %s", err)
+		}
+		hash, err := s.add(ctx, issuanceChain)
+		if err != nil {
+			return &trillian.LogLeaf{}, fmt.Errorf("failed to add issuance chain into CTFE storage: %s", err)
+		}
+		leaf, err := util.BuildLogLeafWithChainHash(logPrefix, *merkleLeaf, 0, raw[0], hash, isPrecert)
+		if err != nil {
+			return &trillian.LogLeaf{}, fmt.Errorf("failed to build LogLeaf: %s", err)
+		}
+		return &leaf, nil
+	} else {
+		leaf, err := util.BuildLogLeaf(logPrefix, *merkleLeaf, 0, raw[0], raw[1:], isPrecert)
+		if err != nil {
+			return &trillian.LogLeaf{}, fmt.Errorf("failed to build LogLeaf: %s", err)
+		}
+		return &leaf, nil
+	}
+}
+
+// FixLogLeaf recreates and populates the LogLeaf.ExtraData if CTFE storage
+// backend is enabled and the type of LogLeaf.ExtraData contains any hash
+// (e.g. PrecertChainEntryHash, CertificateChainHash).
+func (s *issuanceChainService) FixLogLeaf(ctx context.Context, leaf *trillian.LogLeaf) error {
+	// Skip if CTFE storage backend is not enabled.
+	if !s.isCTFEStorageEnabled() {
+		return nil
+	}
+
+	// As the struct stored in leaf.ExtraData is unknown, the only way is to try to unmarshal with each possible struct.
+	// Try to unmarshal with ct.PrecertChainEntryHash struct.
+	var precertChainHash ct.PrecertChainEntryHash
+	if rest, err := tls.Unmarshal(leaf.ExtraData, &precertChainHash); err == nil && len(rest) == 0 {
+		var chain []ct.ASN1Cert
+		if len(precertChainHash.IssuanceChainHash) > 0 {
+			chainBytes, err := s.GetByHash(ctx, precertChainHash.IssuanceChainHash)
+			if err != nil {
+				return err
+			}
+
+			if rest, err := asn1.Unmarshal(chainBytes, &chain); err != nil {
+				return err
+			} else if len(rest) > 0 {
+				return fmt.Errorf("IssuanceChain: trailing data %d bytes", len(rest))
+			}
+		}
+
+		precertChain := ct.PrecertChainEntry{
+			PreCertificate:   precertChainHash.PreCertificate,
+			CertificateChain: chain,
+		}
+		extraData, err := tls.Marshal(precertChain)
+		if err != nil {
+			return err
+		}
+
+		leaf.ExtraData = extraData
+		return nil
+	}
+
+	// Try to unmarshal with ct.CertificateChainHash struct.
+	var certChainHash ct.CertificateChainHash
+	if rest, err := tls.Unmarshal(leaf.ExtraData, &certChainHash); err == nil && len(rest) == 0 {
+		var entries []ct.ASN1Cert
+		if len(certChainHash.IssuanceChainHash) > 0 {
+			chainBytes, err := s.GetByHash(ctx, certChainHash.IssuanceChainHash)
+			if err != nil {
+				return err
+			}
+
+			if rest, err := asn1.Unmarshal(chainBytes, &entries); err != nil {
+				return err
+			} else if len(rest) > 0 {
+				return fmt.Errorf("IssuanceChain: trailing data %d bytes", len(rest))
+			}
+		}
+
+		certChain := ct.CertificateChain{
+			Entries: entries,
+		}
+		extraData, err := tls.Marshal(certChain)
+		if err != nil {
+			return err
+		}
+
+		leaf.ExtraData = extraData
+		return nil
+	}
+
+	// Skip if the types are ct.PrecertChainEntry or ct.CertificateChain as there is no hash.
+	var precertChain ct.PrecertChainEntry
+	if rest, err := tls.Unmarshal(leaf.ExtraData, &precertChain); err == nil && len(rest) == 0 {
+		return nil
+	}
+	var certChain ct.CertificateChain
+	if rest, err := tls.Unmarshal(leaf.ExtraData, &certChain); err == nil && len(rest) == 0 {
+		return nil
+	}
+
+	return fmt.Errorf("unknown extra data type in log leaf: %s", string(leaf.MerkleLeafHash))
+}
+
 // issuanceChainHash returns the SHA-256 hash of the chain.
 func issuanceChainHash(chain []byte) []byte {
 	checksum := sha256.Sum256(chain)
diff --git a/trillian/ctfe/services_test.go b/trillian/ctfe/services_test.go
index 3a9f7c12b4..4082fc4bff 100644
--- a/trillian/ctfe/services_test.go
+++ b/trillian/ctfe/services_test.go
@@ -39,7 +39,7 @@ func TestIssuanceChainServiceAddAndGet(t *testing.T) {
 	issuanceChainService := newIssuanceChainService(storage, cache)
 
 	for _, test := range tests {
-		hash, err := issuanceChainService.Add(ctx, test.chain)
+		hash, err := issuanceChainService.add(ctx, test.chain)
 		if err != nil {
 			t.Errorf("IssuanceChainService.Add(): %v", err)
 		}
diff --git a/trillian/ctfe/storage/storage.go b/trillian/ctfe/storage/storage.go
index 5fccb72419..d9334cbc6b 100644
--- a/trillian/ctfe/storage/storage.go
+++ b/trillian/ctfe/storage/storage.go
@@ -17,6 +17,11 @@ package storage
 
 import (
 	"context"
+	"errors"
+	"strings"
+
+	"github.com/google/certificate-transparency-go/trillian/ctfe/configpb"
+	"github.com/google/certificate-transparency-go/trillian/ctfe/storage/mysql"
 )
 
 // IssuanceChainStorage is an interface which allows CTFE binaries to use different storage implementations for issuance chains.
@@ -27,3 +32,19 @@ type IssuanceChainStorage interface {
 	// Add inserts the key-value pair of issuance chain.
 	Add(ctx context.Context, key []byte, chain []byte) error
 }
+
+// NewIssuanceChainStorage returns nil for Trillian gRPC or mysql.IssuanceChainStorage when MySQL is the prefix in database connection string.
+func NewIssuanceChainStorage(ctx context.Context, backend configpb.LogConfig_IssuanceChainStorageBackend, dbConn string) (IssuanceChainStorage, error) {
+	switch backend {
+	case configpb.LogConfig_ISSUANCE_CHAIN_STORAGE_BACKEND_TRILLIAN_GRPC:
+		return nil, nil
+	case configpb.LogConfig_ISSUANCE_CHAIN_STORAGE_BACKEND_CTFE:
+		if strings.HasPrefix(dbConn, "mysql") {
+			return mysql.NewIssuanceChainStorage(ctx, dbConn), nil
+		} else {
+			return nil, errors.New("failed to initialise IssuanceChainService due to unsupported driver in CTFE storage connection string")
+		}
+	}
+
+	return nil, errors.New("unsupported issuance chain storage backend")
+}
diff --git a/trillian/examples/deployment/docker/ctfe/README.md b/trillian/examples/deployment/docker/ctfe/README.md
index ab5d2bff34..c3953433ce 100644
--- a/trillian/examples/deployment/docker/ctfe/README.md
+++ b/trillian/examples/deployment/docker/ctfe/README.md
@@ -39,6 +39,7 @@ First bring up the trillian instance and the database:
 # Terminal 1
 cd ${GIT_HOME}/certificate-transparency-go/trillian/examples/deployment/docker/ctfe/
 docker compose up
+docker exec -i ctfe-db mariadb -pzaphod -Dtest < ${GIT_HOME}/certificate-transparency-go/trillian/ctfe/storage/mysql/schema.sql
 ```
 
 This brings up everything except the CTFE. Now to provision the logs.
diff --git a/trillian/examples/deployment/docker/ctfe/ct_server.cfg b/trillian/examples/deployment/docker/ctfe/ct_server.cfg
index d16998f261..213b6c565d 100644
--- a/trillian/examples/deployment/docker/ctfe/ct_server.cfg
+++ b/trillian/examples/deployment/docker/ctfe/ct_server.cfg
@@ -12,4 +12,6 @@ config {
 	}
 	max_merge_delay_sec: 86400
 	expected_merge_delay_sec: 120
+	ctfe_storage_connection_string: "mysql://test:zaphod@tcp(localhost:3306)/test"
+	extra_data_issuance_chain_storage_backend: ISSUANCE_CHAIN_STORAGE_BACKEND_CTFE
 }
diff --git a/trillian/examples/deployment/docker/ctfe/docker-compose.yaml b/trillian/examples/deployment/docker/ctfe/docker-compose.yaml
index e2db6c2d0c..2e06abb89d 100644
--- a/trillian/examples/deployment/docker/ctfe/docker-compose.yaml
+++ b/trillian/examples/deployment/docker/ctfe/docker-compose.yaml
@@ -66,6 +66,7 @@ services:
     volumes:
       - ctfe_config:/ctfe-config:ro
     depends_on:
+      - db
       - trillian-log-server
 
 volumes:
diff --git a/trillian/integration/ct_integration_test.cfg b/trillian/integration/ct_integration_test.cfg
index dcdbd206a9..f94d9474fb 100644
--- a/trillian/integration/ct_integration_test.cfg
+++ b/trillian/integration/ct_integration_test.cfg
@@ -28,6 +28,7 @@ config {
 	}
 	max_merge_delay_sec: 86400
 	expected_merge_delay_sec: 120
+	extra_data_issuance_chain_storage_backend: ISSUANCE_CHAIN_STORAGE_BACKEND_TRILLIAN_GRPC
 }
 config {
 	log_id: @TREE_ID@
@@ -43,4 +44,6 @@ config {
 	}
 	max_merge_delay_sec: 86400
 	expected_merge_delay_sec: 120
+	ctfe_storage_connection_string: "mysql://cttest:beeblebrox@tcp(mysql:3306)/cttest"
+	extra_data_issuance_chain_storage_backend: ISSUANCE_CHAIN_STORAGE_BACKEND_CTFE
 }
diff --git a/trillian/integration/ct_lifecycle_test.cfg b/trillian/integration/ct_lifecycle_test.cfg
index c0d704c7d1..e845b2c489 100644
--- a/trillian/integration/ct_lifecycle_test.cfg
+++ b/trillian/integration/ct_lifecycle_test.cfg
@@ -27,6 +27,7 @@ config {
 	}
 	max_merge_delay_sec: 86400
 	expected_merge_delay_sec: 120
+	extra_data_issuance_chain_storage_backend: ISSUANCE_CHAIN_STORAGE_BACKEND_TRILLIAN_GRPC
 }
 config {
 	log_id: @TREE_ID@
@@ -42,4 +43,6 @@ config {
 	}
 	max_merge_delay_sec: 86400
 	expected_merge_delay_sec: 120
+	ctfe_storage_connection_string: "mysql://cttest:beeblebrox@tcp(mysql:3306)/cttest"
+	extra_data_issuance_chain_storage_backend: ISSUANCE_CHAIN_STORAGE_BACKEND_CTFE
 }
diff --git a/trillian/util/log_leaf.go b/trillian/util/log_leaf.go
index 71d3d89b69..6673195016 100644
--- a/trillian/util/log_leaf.go
+++ b/trillian/util/log_leaf.go
@@ -29,28 +29,7 @@ func BuildLogLeaf(logPrefix string,
 	merkleLeaf ct.MerkleTreeLeaf, leafIndex int64,
 	cert ct.ASN1Cert, chain []ct.ASN1Cert, isPrecert bool,
 ) (trillian.LogLeaf, error) {
-	leafData, err := tls.Marshal(merkleLeaf)
-	if err != nil {
-		klog.Warningf("%s: Failed to serialize Merkle leaf: %v", logPrefix, err)
-		return trillian.LogLeaf{}, err
-	}
-
-	extraData, err := ExtraDataForChain(cert, chain, isPrecert)
-	if err != nil {
-		klog.Warningf("%s: Failed to serialize chain for ExtraData: %v", logPrefix, err)
-		return trillian.LogLeaf{}, err
-	}
-
-	// leafIDHash allows Trillian to detect duplicate entries, so this should be
-	// a hash over the cert data.
-	leafIDHash := sha256.Sum256(cert.Data)
-
-	return trillian.LogLeaf{
-		LeafValue:        leafData,
-		ExtraData:        extraData,
-		LeafIndex:        leafIndex,
-		LeafIdentityHash: leafIDHash[:],
-	}, nil
+	return buildLogLeaf(logPrefix, merkleLeaf, leafIndex, cert, chain, nil, isPrecert)
 }
 
 // ExtraDataForChain creates the extra data associated with a log entry as
@@ -71,3 +50,60 @@ func ExtraDataForChain(cert ct.ASN1Cert, chain []ct.ASN1Cert, isPrecert bool) ([
 	}
 	return tls.Marshal(extra)
 }
+
+func BuildLogLeafWithChainHash(logPrefix string, merkleLeaf ct.MerkleTreeLeaf, leafIndex int64, cert ct.ASN1Cert, chainHash []byte, isPrecert bool) (trillian.LogLeaf, error) {
+	return buildLogLeaf(logPrefix, merkleLeaf, leafIndex, cert, nil, chainHash, isPrecert)
+}
+
+// ExtraDataForChainHash creates the extra data associated with a log entry as
+// described in RFC6962 section 4.6 except the chain being replaced with its hash.
+func ExtraDataForChainHash(cert ct.ASN1Cert, chainHash []byte, isPrecert bool) ([]byte, error) {
+	var extra interface{}
+
+	if isPrecert {
+		// For a pre-cert, the extra data is a TLS-encoded PrecertChainEntry.
+		extra = ct.PrecertChainEntryHash{
+			PreCertificate:    cert,
+			IssuanceChainHash: chainHash,
+		}
+	} else {
+		// For a certificate, the extra data is a TLS-encoded:
+		//   ASN.1Cert certificate_chain<0..2^24-1>;
+		// containing the chain after the leaf.
+		extra = ct.CertificateChainHash{
+			IssuanceChainHash: chainHash,
+		}
+	}
+	return tls.Marshal(extra)
+}
+
+// buildLogLeaf builds the trillian.LogLeaf. The chainHash argument controls
+// whether ExtraDataForChain or ExtraDataForChainHash method will be called.
+// If chainHash is not nil, but neither is chain, then chain will be ignored.
+func buildLogLeaf(logPrefix string, merkleLeaf ct.MerkleTreeLeaf, leafIndex int64, cert ct.ASN1Cert, chain []ct.ASN1Cert, chainHash []byte, isPrecert bool) (trillian.LogLeaf, error) {
+	leafData, err := tls.Marshal(merkleLeaf)
+	if err != nil {
+		klog.Warningf("%s: Failed to serialize Merkle leaf: %v", logPrefix, err)
+		return trillian.LogLeaf{}, err
+	}
+
+	var extraData []byte
+	if chainHash == nil {
+		extraData, err = ExtraDataForChain(cert, chain, isPrecert)
+	} else {
+		extraData, err = ExtraDataForChainHash(cert, chainHash, isPrecert)
+	}
+	if err != nil {
+		klog.Warningf("%s: Failed to serialize chain for ExtraData: %v", logPrefix, err)
+		return trillian.LogLeaf{}, err
+	}
+	// leafIDHash allows Trillian to detect duplicate entries, so this should be
+	// a hash over the cert data.
+	leafIDHash := sha256.Sum256(cert.Data)
+	return trillian.LogLeaf{
+		LeafValue:        leafData,
+		ExtraData:        extraData,
+		LeafIndex:        leafIndex,
+		LeafIdentityHash: leafIDHash[:],
+	}, nil
+}